Hadoop RPC框架

??????? serverChannel.register(selector, SelectionKey.OP_ACCEPT);
??? }

??? /**
???? * 採用轮询的方式监听selector上是否有须要处理的事件。假设有，则进行处理
???? * @throws IOException
???? */
??? @SuppressWarnings("unchecked")
??? public void listen() throws IOException {
??????? System.out.println("服务端启动成功！

");
??????? // 轮询訪问selector
??????? while (true) {
??????????? //当注冊的事件到达时，方法返回。否则,该方法会一直堵塞
????????????selector.select();
??????????? // 获得selector中选中的项的迭代器，选中的项为注冊的事件
??????????? Iterator ite = this.selector.selectedKeys().iterator();
??????????? while (ite.hasNext()) {
??????????????? SelectionKey key = (SelectionKey) ite.next();
??????????????? // 删除已选的key,以防反复处理
??????????????? ite.remove();
??????????????? // client请求连接事件
??????????????? if (key.isAcceptable()) {
??????????????????? ServerSocketChannel server = (ServerSocketChannel) key
??????????????????????????? .channel();
??????????????????? // 获得和client连接的通道
??????????????????? SocketChannel channel = server.accept();
??????????????????? // 设置成非堵塞
??????????????????? channel.configureBlocking(false);

??????????????????? //在这里能够给client发送信息哦
??????????????????? channel.write(ByteBuffer.wrap(new String("向client发送了一条信息").getBytes()));
??????????????????? //在和client连接成功之后。为了能够接收到client的信息，须要给通道设置读的权限。
????????????????????channel.register(this.selector, SelectionKey.OP_READ);

??????????????????? // 获得了可读的事件
??????????????? } else if (key.isReadable()) {
??????????????????? read(key);
??????????????? }

??????????? }

??????? }
??? }
??? /**
???? * 处理读取client发来的信息 的事件
???? * @param key
???? * @throws IOException
???? */
??? public void read(SelectionKey key) throws IOException{
??????? // server可读取消息:得到事件发生的Socket通道
??????? SocketChannel channel = (SocketChannel) key.channel();
??????? // 创建读取的缓冲区
??????? ByteBuffer buffer = ByteBuffer.allocate(10);
??????? channel.read(buffer);
??????? byte[] data = buffer.array();
??????? String msg = new String(data).trim();
??????? System.out.println("服务端收到信息："+msg);
??????? ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes());
??????? channel.write(outBuffer);// 将消息回送给client
??? }

??? /**
???? * 启动服务端測试
???? * @throws IOException
???? */
??? public static void main(String[] args) throws IOException {
??????? NIOServer server = new NIOServer();
??????? server.initServer(8000);
??????? server.listen();
??? }
}



package com.sohu.tv.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

/**
* NIOclient
* @author 小路
*/
public class NIOClient {
??? //通道管理器
??? private Selector selector;

??? /**
???? * 获得一个Socket通道，并对该通道做一些初始化的工作
???? * @param ip 连接的server的ip
???? * @param port? 连接的server的端口号
???? * @throws IOException
???? */
??? public void initClient(String ip,int port) throws IOException {
??????? // 获得一个Socket通道
??????? SocketChannel channel = SocketChannel.open();
??????? // 设置通道为非堵塞
??????? channel.configureBlocking(false);
??????? // 获得一个通道管理器
??????? this.selector = Selector.open();

??????? // client连接server,事实上方法运行并没有实现连接，须要在listen（）方法中调
??????? //用channel.finishConnect();才干完毕连接
??????? channel.connect(new InetSocketAddress(ip,port));
??????? //将通道管理器和该通道绑定，并为该通道注冊SelectionKey.OP_CONNECT事件。

??????? channel.register(selector, SelectionKey.OP_CONNECT);
??? }

??? /**
???? * 採用轮询的方式监听selector上是否有须要处理的事件。假设有。则进行处理
???? * @throws IOException
???? */
??? @SuppressWarnings("unchecked")
??? public void listen() throws IOException {
??????? // 轮询訪问selector
??????? while (true) {
??????????? selector.select();
??????????? // 获得selector中选中的项的迭代器
??????????? Iterator ite = this.selector.selectedKeys().iterator();
??????????? while (ite.hasNext()) {
??????????????? SelectionKey key = (SelectionKey) ite.next();
??????????????? // 删除已选的key,以防反复处理
??????????????? ite.remove();
??????????????? // 连接事件发生
??????????????? if (key.isConnectable()) {
??????????????????? SocketChannel channel = (SocketChannel) key
??????????????????????????? .channel();
??????????????????? // 假设正在连接，则完毕连接
??????????????????? if(channel.isConnectionPending()){
??????????????????????? channel.finishConnect();

??????????????????? }
??????????????????? // 设置成非堵塞
??????????????????? channel.configureBlocking(false);

??????????????????? //在这里能够给服务端发送信息哦
??????????????????? channel.write(ByteBuffer.wrap(new String("向服务端发送了一条信息").getBytes()));
??????????????????? //在和服务端连接成功之后，为了能够接收到服务端的信息，须要给通道设置读的权限。

??????????????????? channel.register(this.selector, SelectionKey.OP_READ);

??????????????????? // 获得了可读的事件
??????????????? } else if (key.isReadable()) {
??????????????????? read(key);
??????????????? }
??????????? }
??????? }
??? }
??? /**
???? * 处理读取服务端发来的信息 的事件
???? * @param key
???? * @throws IOException
???? */
??? public void read(SelectionKey key) throws IOException{
??????? //和服务端的read方法一样
??? }


??? /**
???? * 启动client測试
???? * @throws IOException
???? */
??? public static void main(String[] args) throws IOException {
??????? NIOClient client = new NIOClient();
??????? client.initClient("localhost",8000);
??????? client.listen();
??? }

}