在Java中使用NIO進行網絡編程

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中找到需要服務的實例。

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. }  

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. }  

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);