一.
分布式rpc框架有很多，比如dubbo,netty,還有很多其他的產品。但他們大部分都是基於nio的，
nio是非阻塞的io,那么它的內部機制是怎么實現的呢。
1.由一個專門的線程處理所有IO事件，並負責分發。
2.事件驅動機制，事件到來的時候觸發操作，不需要阻塞的監視事件。
3.線程之前通過wait,notify通信，減少線程切換。
上圖是nio的通信模型。
其中：
服務端和客戶端各自維護一個管理通道的對象，我們稱之為selector，它可以監控一個或多個通道上的事件。
采用了雙向通道（channel）進行數據傳輸，而不是單向的流（stream），在通道上可以注冊我們感興趣的事件。
二.
nio server端簡單實現：
package com.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.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
/**
* Created by viruser on 2017/6/8.
*/
public class NioServer {
// 通道管理器
private Selector selector;
/**
* 獲得一個ServerSocket通道，並對該通道做一些初始化的工作
* @param port 綁定的端口號
* @throws IOException
*/
public void initServer(int port) throws IOException {
// 獲得一個ServerSocketChannel通道
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// 設置通道為非阻塞
serverSocketChannel.configureBlocking(false);
// 將該通道對應的ServerSocket綁定到port端口
serverSocketChannel.bind(new InetSocketAddress(port));
// 獲得一個通道管理器
this.selector = Selector.open();
// 將通道管理器和該通道綁定，並為該通道注冊SelectionKey.OP_ACCEPT事件,注冊該事件后，
// 當該事件到達時，selector.select()會返回，如果該事件沒到達selector.select()會一直阻塞。
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
}
/**
* 采用輪詢的方式監聽selector上是否有需要處理的事件，如果有，則進行處理
* @throws IOException
*/
public void listen() throws IOException {
System.out.println("服務端啟動成功！");
// 輪詢訪問selector
while (true) {
// 當注冊的事件到達時，方法返回；否則,該方法會一直阻塞
selector.select();
// 獲得selector中選中的項的迭代器，選中的項為注冊的事件
Iterator<SelectionKey> ite = this.selector.selectedKeys().iterator();
while (ite.hasNext()) {
SelectionKey key = (SelectionKey) ite.next();
// 刪除已選的key,以防重復處理
ite.remove();
if (key.isAcceptable()) {// 客戶端請求連接事件
ServerSocketChannel server = (ServerSocketChannel) key.channel();
// 獲得和客戶端連接的通道
SocketChannel channel = server.accept();
// 設置成非阻塞
channel.configureBlocking(false);
// 在這里可以給客戶端發送信息哦
channel.write(ByteBuffer.wrap(new String("向客戶端發送了一條信息")
.getBytes("utf-8")));
// 在和客戶端連接成功之后，為了可以接收到客戶端的信息，需要給通道設置讀的權限。
channel.register(this.selector, SelectionKey.OP_READ);
} else if (key.isReadable()) {// 獲得了可讀的事件
read(key);
}
}
}
}
/**
* 處理讀取客戶端發來的信息 的事件
*
* @param key
* @throws IOException
*/
public void read(SelectionKey key) throws IOException {
// 服務器可讀取消息:得到事件發生的Socket通道
SocketChannel channel = (SocketChannel) key.channel();
// 創建讀取的緩沖區
ByteBuffer buffer = ByteBuffer.allocate(512);
channel.read(buffer);
byte[] data = buffer.array();
String msg = new String(data).trim();
System.out.println("服務端收到信息：" + msg);
ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes("utf-8"));
channel.write(outBuffer);// 將消息回送給客戶端
}
/**
* 啟動服務端測試
*
* @throws IOException
*/
public static void main(String[] args) throws IOException {
NioServer server = new NioServer();
server.initServer(8000);
server.listen();
}
}
客戶端：
package com.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;
/**
* Created by viruser on 2017/6/8.
*/
public class NioClient {
//通道管理器
private Selector selector;
/**
* 獲得一個Socket通道，並對該通道做一些初始化的工作
* @param ip 連接的服務器的ip
* @param port 連接的服務器的端口號
* @throws IOException
*/
public void initClient(String ip,int port) throws IOException {
// 獲得一個Socket通道
SocketChannel channel = SocketChannel.open();
// 設置通道為非阻塞
channel.configureBlocking(false);
// 獲得一個通道管理器
this.selector = Selector.open();
// 客戶端連接服務器,其實方法執行並沒有實現連接，需要在listen（）方法中調
//用channel.finishConnect();才能完成連接
channel.connect(new InetSocketAddress(ip,port));
//將通道管理器和該通道綁定，並為該通道注冊SelectionKey.OP_CONNECT事件。
channel.register(selector, SelectionKey.OP_CONNECT);
}
/**
* 采用輪詢的方式監聽selector上是否有需要處理的事件，如果有，則進行處理
* @throws IOException
*/
public void connect() throws IOException {
// 輪詢訪問selector
while (true) {
// 選擇一組可以進行I/O操作的事件，放在selector中,客戶端的該方法不會阻塞，
//這里和服務端的方法不一樣，查看api注釋可以知道，當至少一個通道被選中時，
//selector的wakeup方法被調用，方法返回，而對於客戶端來說，通道一直是被選中的
selector.select();
// 獲得selector中選中的項的迭代器
Iterator<SelectionKey> 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("utf-8")));
//在和服務端連接成功之后，為了可以接收到服務端的信息，需要給通道設置讀的權限。
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方法一樣
// 服務器可讀取消息:得到事件發生的Socket通道
SocketChannel channel = (SocketChannel) key.channel();
// 創建讀取的緩沖區
ByteBuffer buffer = ByteBuffer.allocate(512);
channel.read(buffer);
byte[] data = buffer.array();
String msg = new String(data).trim();
System.out.println("客戶端收到信息：" + msg);
ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes("utf-8"));
channel.write(outBuffer);// 將消息回送給客戶端
}
/**
* 啟動客戶端測試
* @throws IOException
*/
public static void main(String[] args) throws IOException {
NioClient client = new NioClient();
client.initClient("localhost",8000);
client.connect();
}
}