利用 Java NIO 非阻塞IO流实现高性能服务器

使用Java NIO编写高性能的服务器

一个常见的网络IO通讯流程如下:



图3：网络通讯基本过程
从该网络通讯过程来理解一下何为阻塞:
在以上过程中若连接还没到来，那么accept会阻塞,程序运行到这里不得不挂起，CPU转而执行其他线程。
在以上过程中若数据还没准备好，read会一样也会阻塞。
阻塞式网络IO的特点：多线程处理多个连接。每个线程拥有自己的栈空间并且占用一些CPU时间。每个线程遇到外部为准备好的时候，都会阻塞掉。阻塞的结果就是会带来大量的进程上下文切换。且大部分进程上下文切换可能是无意义的。比如假设一个线程监听一个端口，一天只会有几次请求进来，但是该cpu不得不为该线程不断做上下文切换尝试，大部分的切换以阻塞告终。
非阻塞的原理
把整个过程切换成小的任务，通过任务间协作完成。
由一个专门的线程来处理所有的IO事件，并负责分发。
事件驱动机制：事件到的时候触发，而不是同步的去监视事件。
线程通讯：线程之间通过wait,notify等方式通讯。保证每次上下文切换都是有意义的。减少无谓的进程切换。

NIO模式的基本原理描述如下：
服务端打开一个通道（ServerSocketChannel），并向通道中注册一个选择器（Selector），这个选择器是与一些感兴趣的操作的标识（SelectionKey，即通过这个标识可以定位到具体的操作，从而进行响应的处理）相关联的，然后基于选择器（Selector）轮询通道（ServerSocketChannel）上注册的事件，并进行相应的处理。
客户端在请求与服务端通信时，也可以向服务器端一样注册（比服务端少了一个SelectionKey.OP_ACCEPT操作集合），并通过轮询来处理指定的事件，而不必阻塞。
下面的例子，主要以服务端为例，而客户端只是简单地发送请求数据和读响应数据。
服务端实现，代码如下所示：
1. packageorg.shirdrn.java.communications.nio;
2.
3. importjava.io.IOException;
4. importjava.net.InetSocketAddress;
5. importjava.nio.ByteBuffer;
6. importjava.nio.channels.SelectionKey;
7. importjava.nio.channels.Selector;
8. importjava.nio.channels.ServerSocketChannel;
9. importjava.nio.channels.SocketChannel;
10. importjava.util.Iterator;
11. importjava.util.Set;
12. importjava.util.logging.Logger;
13.
14. /**
15. * NIO服务端
16. *
17. * @authorshirdrn
18. */
19. public class NioTcpServer extends Thread {
20.
21. private static final Logger log =Logger.getLogger(NioTcpServer.class.getName());
22. privateInetSocketAddress inetSocketAddress;
23. privateHandler handler = newServerHandler();
24.
25. publicNioTcpServer(String hostname, intport) {
26. inetSocketAddress = new InetSocketAddress(hostname, port);
27. }
28.
29. @Override
30. public void run() {
31. try {
32. Selector selector = Selector.open();// 打开选择器
33. ServerSocketChannelserverSocketChannel = ServerSocketChannel.open(); // 打开通道
34. serverSocketChannel.configureBlocking(false); // 非阻塞
35. serverSocketChannel.socket().bind(inetSocketAddress);
36. serverSocketChannel.register(selector,SelectionKey.OP_ACCEPT); // 向通道注册选择器和对应事件标识
37. log.info("Server: socket server started.");
38. while(true) { // 轮询
39. intnKeys = selector.select();
40. if(nKeys>0) {
41. Set<SelectionKey> selectedKeys= selector.selectedKeys();
42. Iterator<SelectionKey> it =selectedKeys.iterator();
43. while(it.hasNext()){
44. SelectionKey key = it.next();
45. if(key.isAcceptable()){
46. log.info("Server: SelectionKey is acceptable.");
47. handler.handleAccept(key);
48. } else if(key.isReadable()) {
49. log.info("Server: SelectionKey is readable.");
50. handler.handleRead(key);
51. } else if(key.isWritable()) {
52. log.info("Server: SelectionKey is writable.");
53. handler.handleWrite(key);
54. }
55. it.remove();
56. }
57. }
58. }
59. } catch (IOException e) {
60. e.printStackTrace();
61. }
62. }
63.
64. /**
65. * 简单处理器接口
66. *
67. * @authorshirdrn
68. */
69. interfaceHandler {
70. /**
71. * 处理{@linkSelectionKey#OP_ACCEPT}事件
72. * @paramkey
73. * @throwsIOException
74. */
75. voidhandleAccept(SelectionKey key) throwsIOException;
76. /**
77. * 处理{@linkSelectionKey#OP_READ}事件
78. * @paramkey
79. * @throwsIOException
80. */
81. voidhandleRead(SelectionKey key) throwsIOException;
82. /**
83. * 处理{@linkSelectionKey#OP_WRITE}事件
84. * @paramkey
85. * @throwsIOException
86. */
87. voidhandleWrite(SelectionKey key) throwsIOException;
88. }
89.
90. /**
91. * 服务端事件处理实现类
92. *
93. * @authorshirdrn
94. */
95. classServerHandler implementsHandler {
96.
97. @Override
98. public void handleAccept(SelectionKey key) throws IOException {
99. ServerSocketChannelserverSocketChannel = (ServerSocketChannel)key.channel();
100.SocketChannel socketChannel =serverSocketChannel.accept();
101.log.info("Server: accept client socket " +socketChannel);
102.socketChannel.configureBlocking(false);
103.socketChannel.register(key.selector(),SelectionKey.OP_READ);
104.}
105.
106.@Override
107.public void handleRead(SelectionKey key) throws IOException {
108.ByteBuffer byteBuffer =ByteBuffer.allocate(512);
109.SocketChannel socketChannel =(SocketChannel)key.channel();
110.while(true) {
111.intreadBytes = socketChannel.read(byteBuffer);
112.if(readBytes>0) {
113.log.info("Server: readBytes = " +readBytes);
114.log.info("Server: data = " + new String(byteBuffer.array(), 0, readBytes));
115.byteBuffer.flip();
116.socketChannel.write(byteBuffer);
117.break;
118.}
119.}
120.socketChannel.close();
121.}
122.
123.@Override
124.public void handleWrite(SelectionKey key) throws IOException {
125.ByteBuffer byteBuffer = (ByteBuffer)key.attachment();
126.byteBuffer.flip();
127.SocketChannel socketChannel =(SocketChannel)key.channel();
128.socketChannel.write(byteBuffer);
129.if(byteBuffer.hasRemaining()){
130.key.interestOps(SelectionKey.OP_READ);
131.}
132.byteBuffer.compact();
133.}
134.}
135.
136.public static void main(String[] args) {
137.NioTcpServer server = new NioTcpServer("localhost", 1000);
138.server.start();
139.}
140.}
客户端实现，代码如下所示：

1. packageorg.shirdrn.java.communications.nio;
2.
3. import java.io.IOException;
4. importjava.net.InetSocketAddress;
5. import java.nio.ByteBuffer;
6. importjava.nio.channels.SocketChannel;
7. importjava.util.logging.Logger;
8.
9. /**
10. *NIO客户端
11. *
12. *@author shirdrn
13. */
14. public classNioTcpClient {
15.
16. private staticfinal Logger log = Logger.getLogger(NioTcpClient.class.getName());
17. private InetSocketAddressinetSocketAddress;
18.
19. public NioTcpClient(Stringhostname, int port) {
20. inetSocketAddress = new InetSocketAddress(hostname, port);
21. }
22.
23. /**
24. * 发送请求数据
25. *@param requestData
26. */
27. public voidsend(String requestData) {
28. try {
29. SocketChannelsocketChannel = SocketChannel.open(inetSocketAddress);
30. socketChannel.configureBlocking(false);
31. ByteBuffer byteBuffer =ByteBuffer.allocate(512);
32. socketChannel.write(ByteBuffer.wrap(requestData.getBytes()));
33. while (true){
34. byteBuffer.clear();
35. int readBytes =socketChannel.read(byteBuffer);
36. if (readBytes > 0) {
37. byteBuffer.flip();
38. log.info("Client: readBytes = " + readBytes);
39. log.info("Client: data = " + newString(byteBuffer.array(), 0, readBytes));
40. socketChannel.close();
41. break;
42. }
43. }
44.
45. } catch(IOException e) {
46. e.printStackTrace();
47. }
48. }
49.
50. public staticvoid main(String[] args) {
51. String hostname = "localhost";
52. String requestData = "Actions speak louder than words!";
53. int port = 1000;
54. new NioTcpClient(hostname,port).send(requestData);
55. }
56. }
客户端和服务端可以采用同样轮询的非阻塞模式来实现，为简单实现在这个例子中我们把客户端角色简化了，而实际上它可能在另一个系统通信中充当服务端角色。
另外，上面对于不同事件是采用非线程的方式来处理，只是简单地调用处理的方法。在实际中，如果存在大量连接、读写请求，可以考虑使用线程池来更大程度地并发处理，提高服务端处理的速度和吞吐量，提升系统性能.