在Java中使用NIO进行网络编程

在JDK中，有一个非常有意思的库：NIO（New
I/O）。这个库中有3个重要的类，分别是java.nio.channels中Selector和Channel，以及java.nio中的Buffer。

本篇文章我们首先了解一下为什么需要NIO来进行网络编程，然后看看一步一步来讲解如何在网络编程中使用NIO。

为什么需要NIO

使用Java编写过Socket程序的同学一定都知道Socket和SocketServer。当调用某个调用的时候，调用的地方就会阻塞，等待响应。这种方式对于小规模的程序非常方便，但是对于大型的程序就有点力不从心了，当有大量的连接的时候，我们可以为每一个连接建立一个线程来操作。但是这种做法带来的缺陷也是显而易见的：

1. 硬件能够支持大量的并发。
2. 并发的数量始终有一个上限。
3. 各个线程之间的优先级不好控制。
4. 各个Client之间的交互与同步困难。

我们也可以使用一个线程来处理所有的请求，使用不阻塞的IO，轮询查询所有的Client。这种做法同样也有缺陷：无法迅速响应Client端，同时会消耗大量轮询查询的时间。

所以，我们需要一种poll的模式来处理这种情况，从大量的网络连接中找出来真正需要服务的Client。这正是NIO诞生的原因：提供一种Poll的模式，在所有的Client中找到需要服务的Client。

回到我们刚刚说到的3个最最重要的Class：java.nio.channels中Selector和Channel，以及java.nio中的Buffer。

Channel代表一个可以被用于Poll操作的对象（可以是文件流也可以使网络流），Channel能够被注册到一个Selector中。通过调用Selector的select方法可以从所有的Channel中找到需要服务的实例（Accept，read
..）。

Buffer对象提供读写数据的缓存。相对于我们熟悉的Stream对象，Buffer提供更好的性能以及更好的编程透明性（人为控制缓存的大小以及具体的操作）。

配合BUFFER使用CHANNEL

与传统模式的编程不用，Channel不使用Stream，而是Buffer。我们来实现一个简单的非阻塞Echo Client：

1. package com.cnblogs.gpcuster;
2. import java.net.InetSocketAddress;
3. import java.net.SocketException;
4. import java.nio.ByteBuffer;
5. import java.nio.channels.SocketChannel;
6. public class TCPEchoClientNonblocking {
7. public static void main(String args[]) throws Exception {
8. if ((args.length < 2) || (args.length > 3))// Testforcorrect#ofargs
9. throw new IllegalArgumentException(
10. “Parameter(s): <Server> <Word> [<Port>]”);
11. String server = args[0];// ServernameorIPaddress
12. // ConvertinputStringtobytesusingthedefaultcharset
13. byte[] argument = args[1].getBytes();
14. int servPort = (args.length == 3) ? Integer.parseInt(args[2]) : 7;
15. // Createchannelandsettononblocking
16. SocketChannel clntChan = SocketChannel.open();
17. clntChan.configureBlocking(false);
18. // Initiateconnectiontoserverandrepeatedlypolluntilcomplete
19. if (!clntChan.connect(new InetSocketAddress(server, servPort))) {
20. while (!clntChan.finishConnect()) {
21. System.out.print(“.”);// Dosomethingelse
22. }
23. }
24. ByteBuffer writeBuf = ByteBuffer.wrap(argument);
25. ByteBuffer readBuf = ByteBuffer.allocate(argument.length);
26. int totalBytesRcvd = 0;// Totalbytesreceivedsofar
27. int bytesRcvd;// Bytesreceivedinlastread
28. while (totalBytesRcvd < argument.length) {
29. if (writeBuf.hasRemaining()) {
30. clntChan.write(writeBuf);
31. }
32. if ((bytesRcvd = clntChan.read(readBuf)) == –1) {
33. throw new SocketException(“Connection closed prematurely”);
34. }
35. totalBytesRcvd += bytesRcvd;
36. System.out.print(“.”);// Dosomethingelse
37. }
38. System.out.println(“Received:” + // converttoStringperdefaultcharset
39. new String(readBuf.array(), 0, totalBytesRcvd));
40. clntChan.close();
41. }
42. }

这段代码使用ByteBuffer来保存读写的数据。通过clntChan.configureBlocking(false);
设置后，其中的connect，read，write操作都不回阻塞，而是立刻放回结果。

使用SELECTOR

Selector的可以从所有的被注册到自己Channel中找到需要服务的实例。

我们来实现Echo Server。

首先，定义一个接口：

1. package com.cnblogs.gpcuster;
2. import java.nio.channels.SelectionKey;
3. import java.io.IOException;
4. public interface TCPProtocol {
5. void handleAccept(SelectionKey key) throws IOException;
6. void handleRead(SelectionKey key) throws IOException;
7. void handleWrite(SelectionKey key) throws IOException;
8. }
9. 我们的Echo Server将使用这个接口。然后我们实现Echo Server：
10. import java.io.IOException;
11. import java.net.InetSocketAddress;
12. import java.nio.channels.SelectionKey;
13. import java.nio.channels.Selector;
14. import java.nio.channels.ServerSocketChannel;
15. import java.util.Iterator;
16. public class TCPServerSelector {
17. private static final int BUFSIZE = 256;// Buffersize(bytes)
18. private static final int TIMEOUT = 3000;// Waittimeout(milliseconds)
19. public static void main(String[] args) throws IOException {
20. if (args.length < 1) {// Testforcorrect#ofargs
21. throw new IllegalArgumentException(“Parameter(s):<Port>…”);
22. }
23. // Createaselectortomultiplexlisteningsocketsandconnections
24. Selector selector = Selector.open();
25. // Createlisteningsocketchannelforeachportandregisterselector
26. for (String arg : args) {
27. ServerSocketChannel listnChannel = ServerSocketChannel.open();
28. listnChannel.socket().bind(
29. new InetSocketAddress(Integer.parseInt(arg)));
30. listnChannel.configureBlocking(false);// mustbenonblockingtoregister
31. // Registerselectorwithchannel.Thereturnedkeyisignored
32. listnChannel.register(selector, SelectionKey.OP_ACCEPT);
33. }
34. // Createahandlerthatwillimplementtheprotocol
35. TCPProtocol protocol = new EchoSelectorProtocol(BUFSIZE);
36. while (true) {// Runforever,processingavailableI/Ooperations
37. // Waitforsomechanneltobeready(ortimeout)
38. if (selector.select(TIMEOUT) == 0) {// returns#ofreadychans
39. System.out.print(“.”);
40. continue;
41. }
42. // GetiteratoronsetofkeyswithI/Otoprocess
43. Iterator<SelectionKey> keyIter = selector.selectedKeys().iterator();
44. while (keyIter.hasNext()) {
45. SelectionKey key = keyIter.next();// Keyisbitmask
46. // Serversocketchannelhaspendingconnectionrequests?
47. if (key.isAcceptable()) {
48. protocol.handleAccept(key);
49. }
50. // Clientsocketchannelhaspendingdata?
51. if (key.isReadable()) {
52. protocol.handleRead(key);
53. }
54. // Clientsocketchannelisavailableforwritingand
55. // keyisvalid(i.e.,channelnotclosed)?
56. if (key.isValid() && key.isWritable()) {
57. protocol.handleWrite(key);
58. }
59. keyIter.remove();// removefromsetofselectedkeys
60. }
61. }
62. }
63. }

我们通过listnChannel.register(selector, SelectionKey.OP_ACCEPT);
注册了一个我们感兴趣的事件，然后调用selector.select(TIMEOUT)等待订阅的时间发生，然后再采取相应的处理措施。

最后我们实现EchoSelectorProtocol

1. package com.cnblogs.gpcuster;
2. import java.nio.channels.SelectionKey;
3. import java.nio.channels.SocketChannel;
4. import java.nio.channels.ServerSocketChannel;
5. import java.nio.ByteBuffer;
6. import java.io.IOException;
7. public class EchoSelectorProtocol implements TCPProtocol {
8. private int bufSize;// SizeofI/Obuffer
9. public EchoSelectorProtocol(int bufSize) {
10. this.bufSize = bufSize;
11. }
12. public void handleAccept(SelectionKey key) throws IOException {
13. SocketChannel clntChan = ((ServerSocketChannel) key.channel()).accept();
14. clntChan.configureBlocking(false);// Mustbenonblockingtoregister
15. // Registertheselectorwithnewchannelforreadandattachbytebuffer
16. clntChan.register(key.selector(), SelectionKey.OP_READ, ByteBuffer
17. .allocate(bufSize));
18. }
19. public void handleRead(SelectionKey key) throws IOException {
20. // Clientsocketchannelhaspendingdata
21. SocketChannel clntChan = (SocketChannel) key.channel();
22. ByteBuffer buf = (ByteBuffer) key.attachment();
23. long bytesRead = clntChan.read(buf);
24. if (bytesRead == –1) {// Didtheotherendclose?
25. clntChan.close();
26. } else if (bytesRead > 0) {
27. // Indicateviakeythatreading/writingarebothofinterestnow.
28. key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
29. }
30. }
31. public void handleWrite(SelectionKey key) throws IOException {
32. /*
33. * Channelisavailableforwriting,andkeyisvalid(i.e.,clientchannel
34. * notclosed).
35. */
36. // Retrievedatareadearlier
37. ByteBuffer buf = (ByteBuffer) key.attachment();
38. buf.flip();// Preparebufferforwriting
39. SocketChannel clntChan = (SocketChannel) key.channel();
40. clntChan.write(buf);
41. if (!buf.hasRemaining()) {// Buffercompletelywritten?
42. // Nothingleft,sonolongerinterestedinwrites
43. key.interestOps(SelectionKey.OP_READ);
44. }
45. buf.compact();// Makeroomformoredatatobereadin
46. }
47. }

在这里，我们又进一步对Selector注册了相关的事件：key.interestOps(SelectionKey.OP_READ);

这样，我们就实现了基于NIO的Echo 系统。