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

2021-08-26 07:14:46

在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：

 

[java] view plaincopy
  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。

首先，定义一个接口：

 

[java] view plaincopy
  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. }   

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

[c-sharp] view plaincopy
  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 系统。

上一篇：GoldenGate实时投递数据到大数据平台（1）-MongoDB


下一篇：迪杰斯特拉/dijkstra 算法模板（具体凝视）