java 阻塞模式与非阻塞模式

1. package concurrentTest;  
2.   
3. import java.io.BufferedReader;  
4. import java.io.IOException;  
5. import java.io.InputStreamReader;  
6. import java.io.PrintWriter;  
7. import java.net.ServerSocket;  
8. import java.net.Socket;  
9. import java.net.UnknownHostException;  
10.   
11. /** 
12.  * TCP/IP的阻塞模式  
13.  */  
14. public class TCPBIO {  
15.       
16.     /** 
17.      * 客户端监听事件 
18.      * @param ipStr IP地址 
19.      * @param portNum 端口号 
20.      * */  
21.     public void clientListen(String ipStr, int portNum)  
22.     {  
23.         try {  
24.             Socket socket = new Socket(ipStr, portNum);  
25.               
26.             //创建读取服务器端返回流的BufferReader   
27.             BufferedReader in = new BufferedReader(new   
28.                                     InputStreamReader(socket.getInputStream()));  
29.               
30.             //创建向服务器写入流的PrintWriter   
31.             PrintWriter out = new PrintWriter(socket.getOutputStream(), true);  
32.               
33.             //向服务器发送字符串信息，此处即使写失败也不会抛出异常，会一直阻塞到   
34.             //写入操作系统或者网络IO出现异常为止   
35.             out.println("hello java");  
36.               
37.             //阻塞读取服务器端的返回信息，以下代码会阻塞到服务器端返回信息或者网络IO出现异常为止   
38.             //若希望过段时间后不阻塞，则在socket创建后加入socket.setSoTimeout(time);   
39.             in.readLine();  
40.               
41.         } catch (UnknownHostException e) {  
42.             // TODO Auto-generated catch block   
43.             e.printStackTrace();  
44.         } catch (IOException e) {  
45.             // TODO Auto-generated catch block   
46.             e.printStackTrace();  
47.         }  
48.           
49.     }  
50.       
51.       
52.     /** 
53.      * 服务器端监听事件 
54.      * @param portNum 端口号 
55.      * */  
56.     public void serverListen(int portNum)  
57.     {  
58.         ServerSocket ss;  
59.         try {  
60.             ss = new ServerSocket(portNum);  
61.               
62.             //通过Scoket.getInputStream和Socket.getOutputStream进行读写操作，此方法   
63.             //会一直阻塞到有客户端发送建立请求   
64.             Socket socket = ss.accept();  
65.               
66.         } catch (IOException e) {  
67.             // 网络IO异常   
68.             e.printStackTrace();  
69.         }  
70.           
71.     }  
72.       
73.     public static void main(String[] args) {  
74.         // TODO Auto-generated method stub   
75.   
76.     }  
77.   
78. }  

1. <PRE class=java name="code">package concurrentTest;  
2.   
3. import java.io.IOException;  
4. import java.net.InetSocketAddress;  
5. import java.net.ServerSocket;  
6. import java.nio.ByteBuffer;  
7. import java.nio.channels.*;  
8. import java.util.Set;  
9.   
10. /** 
11.  * TCP/IP的 非阻塞模式  
12.  */  
13.   
14. public class TCPNIO {  
15.   
16.     /** 
17.      * 客户端监听事件 
18.      * @param ipStr IP地址 
19.      * @param portNum 端口号 
20.      * */     
21.       
22.     public void clientListen(String ipStr, int portNum, int blockTime)  
23.     {  
24.         try {  
25.             SocketChannel channel = SocketChannel.open();  
26.               
27.             //设置为非阻塞模式   
28.             channel.configureBlocking(false);  
29.               
30.             //对于非阻塞模式，立刻返回false，表示连接正在建立中   
31.             Selector selector = Selector.open();  
32.               
33.             //向channel注册selector和感兴趣事件   
34.             channel.register(selector, SelectionKey.OP_CONNECT);  
35.               
36.             //阻塞至有感兴趣的IO事件发生，或者到达超时时间，如果希望一直等待到有感兴趣的   
37.             //事件发生，可调用无参数的select方法，如果希望不阻塞直接返回目前是否有感兴趣的   
38.             //事件发生，可调用selectNow方法   
39.             int nKeys = selector.select(blockTime);  
40.             SelectionKey sKey = null;  
41.               
42.             //nKeys大于0，说明有感兴趣的事件发生   
43.             if(nKeys > 0)  
44.             {  
45.                 Set<SelectionKey> keys = selector.selectedKeys();  
46.                 for(SelectionKey key:keys)  
47.                 {  
48.                     //对于发生连接事件   
49.                     if(key.isConnectable())  
50.                     {  
51.                         SocketChannel sc = (SocketChannel)key.channel();  
52.                         sc.configureBlocking(false);  
53.                           
54.                         //注册感兴趣的IO事件，通常不直接注册写事件，在发送缓冲区未满的情况下   
55.                         //一直是可写的，所以如果注册了写事件，而又不写数据，则很容易造成CPU消耗100%   
56.                         sKey = sc.register(selector, SelectionKey.OP_READ);  
57.                           
58.                         //完成连接的建立   
59.                         sc.finishConnect();  
60.                     }  
61.                     else if(key.isReadable())//有流可读取   
62.                     {  
63.                         ByteBuffer buffer = ByteBuffer.allocate(1024);  
64.                         SocketChannel sc = (SocketChannel)key.channel();  
65.                         int readBytes = 0;  
66.   
67.                         try {  
68.                             int ret = 0;  
69.                             try {  
70.                                 //读取目前可读的流，sc.read(buffer)返回的是成功复制到bytebuffer中   
71.                                 //的字节数，为阻塞操作，值可能为0，若到流结尾，返回-1   
72.                                 while ((ret = sc.read(buffer)) > 0) {  
73.                                     readBytes += ret;  
74.                                 }  
75.                             } finally {  
76.                                 buffer.flip();  
77.                             }  
78.                         } finally{  
79.                             if(buffer != null)  
80.                             {  
81.                                 buffer.clear();  
82.                             }  
83.                         }  
84.                     }  
85.                     else if(key.isWritable())//可写入流   
86.                     {  
87.                         //取消对OP_WRITE事件的注册   
88.                         ByteBuffer buffer = ByteBuffer.allocate(1024);  
89.                         key.interestOps(key.interestOps() & (~SelectionKey.OP_WRITE));  
90.                         SocketChannel sc = (SocketChannel)key.channel();  
91.                           
92.                         //此步为阻塞操作，知道写入操作系统发送缓冲区或者网络IO出现异常   
93.                         //返回的为成功写入的字节数，若缓冲区已满，返回0   
94.                         int writeenedSize = sc.write(buffer);  
95.                           
96.                         //若未写入，继续注册感兴趣的OP_WRITE事件   
97.                         if(writeenedSize == 0)  
98.                         {  
99.                             key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);  
100.                         }  
101.                     }  
102.                 }  
103.                   
104.                 selector.selectedKeys().clear();  
105.             }  
106.             /* 
107.              * 也可以直接调用channel.write完成写 
108.              * 只有在写入未成功时才注册OP_WRITE事件 
109.             ByteBuffer buffer = ByteBuffer.allocate(1024); 
110.             int wSize = channel.write(buffer); 
111.             if(wSize == 0) 
112.             { 
113.                 key.interestOps(key.interestOps() | SelectionKey.OP_WRITE); 
114.             }*/  
115.               
116.         } catch (IOException e) {  
117.             // TODO Auto-generated catch block   
118.             e.printStackTrace();  
119.         }  
120.     }  
121.       
122.       
123.     /** 
124.      * 服务器端监听事件 
125.      * @param portNum 端口号 
126.      * */  
127.     public void serverListen(int portNum, int blockTime)  
128.     {  
129.         ServerSocketChannel ssc;  
130.         try {  
131.             ssc = ServerSocketChannel.open();  
132.             ServerSocket serverSocket = ssc.socket();             
133.             serverSocket.bind(new InetSocketAddress(portNum));  
134.             ssc.configureBlocking(false);  
135.             Selector selector = Selector.open();  
136.             ssc.register(selector, SelectionKey.OP_ACCEPT);  
137.               
138.             int nKeys = selector.select(blockTime);           
139.             if(nKeys > 0)  
140.             {  
141.                 Set<SelectionKey> keys = selector.selectedKeys();  
142.                 for(SelectionKey key:keys)  
143.                 {  
144.                     if(key.isAcceptable())  
145.                     {  
146.                         ServerSocketChannel server = (ServerSocketChannel)key.channel();  
147.                         SocketChannel sc = server.accept();  
148.                         if(sc == null)  
149.                         {  
150.                             continue;  
151.                         }  
152.                         sc.configureBlocking(false);  
153.                           
154.                         //注册感兴趣的连接建立事件   
155.                         sc.register(selector, SelectionKey.OP_READ);  
156.                     }  
157.                 }  
158.             }  
159.               
160.         } catch (IOException e) {  
161.             // TODO Auto-generated catch block   
162.             e.printStackTrace();  
163.         }  
164.           
165.     }  
166.       
167.     public static void main(String[] args) {  
168.         // TODO Auto-generated method stub   
169.   
170.     }  
171.   
172. }  
173. </PRE><BR>  
174. <BR>  
175. <PRE></PRE>  
176. <BR>  

1. package concurrentTest;  
2.   
3. import java.io.IOException;  
4. import java.net.DatagramPacket;  
5. import java.net.DatagramSocket;  
6. import java.net.InetAddress;  
7. import java.net.SocketException;  
8.   
9. /** 
10.  * UDP/IP阻塞模式 
11.  * */  
12.   
13. public class UDPBIO {  
14.   
15.     /** 
16.      * 服务器端和客户端类似  
17.      * @param ipStr IP地址 
18.      * @param portNum 端口号 
19.      * */     
20.       
21.     public void udpListen(String ipStr, int portNum)  
22.     {  
23.         try {  
24.             //若希望是全双工模式，则启动一个监听端口，承担服务器的职责   
25.             //若不能绑定到指定端口，则抛出SocketException   
26.             DatagramSocket serverSocket = new DatagramSocket(portNum);  
27.             InetAddress server = InetAddress.getByName(ipStr);  
28.             byte[] buffer = new byte[65507];  
29.             DatagramPacket receivePacket = new DatagramPacket(buffer,buffer.length);  
30.             DatagramSocket socket = new DatagramSocket();  
31.             DatagramPacket packet = new DatagramPacket(buffer,buffer.length,server,portNum);  
32.               
33.             //阻塞发送packet到指定的服务器和端口   
34.             //网络IO异常，抛出IOExceprion   
35.             //连不上IP和端口，抛出PortUnreachableException   
36.             socket.send(packet);  
37.               
38.             //阻塞并同步读取流信息，如接收到的信息流比packet长，则删除更长的信息   
39.             serverSocket.receive(receivePacket);  
40.               
41.         } catch (SocketException e) {  
42.             // 若不能绑定到指定端口   
43.             e.printStackTrace();  
44.         } catch (IOException e) {  
45.             // TODO Auto-generated catch block   
46.             e.printStackTrace();  
47.         }  
48.           
49.     }  
50.       
51.     public static void main(String[] args) {  
52.         // TODO Auto-generated method stub   
53.   
54.     }  
55.   
56. }  

1. package concurrentTest;  
2.   
3. import java.io.IOException;  
4. import java.net.DatagramSocket;  
5. import java.net.InetAddress;  
6. import java.net.InetSocketAddress;  
7. import java.net.MulticastSocket;  
8. import java.net.SocketAddress;  
9. import java.net.SocketException;  
10. import java.net.UnknownHostException;  
11. import java.nio.ByteBuffer;  
12. import java.nio.channels.DatagramChannel;  
13. import java.nio.channels.SelectionKey;  
14. import java.nio.channels.Selector;  
15.   
16.   
17. /** 
18.  * UDP/IP非阻塞模式 
19.  * */  
20. public class UDPNIO {  
21.   
22.     /** 
23.      * 服务器端和客户端类似  
24.      * @param ipStr IP地址 
25.      * @param portNum 端口号 
26.      * */         
27.       
28.     public void udpListen(String ipStr, int portNum)  
29.     {  
30.         try {  
31.             ByteBuffer buffer = ByteBuffer.allocate(1024);  
32.             DatagramChannel reveiveChannel = DatagramChannel.open();  
33.             reveiveChannel.configureBlocking(false);  
34.             DatagramSocket socket = reveiveChannel.socket();  
35.             socket.bind(new InetSocketAddress(portNum));  
36.             Selector selector = Selector.open();  
37.             reveiveChannel.register(selector, SelectionKey.OP_READ);  
38.               
39.             //之后可采取和TCP/IP NIO中对selector遍历方式进行流信息的读取   
40.             DatagramChannel sendChannel = DatagramChannel.open();  
41.             sendChannel.configureBlocking(false);  
42.             SocketAddress target = new InetSocketAddress(InetAddress.getLocalHost(), portNum);  
43.             sendChannel.connect(target);  
44.               
45.             //阻塞写入流，若发送缓冲区已满，返回0，此时可继续注册OP_WRITE事件   
46.             sendChannel.write(buffer);  
47.         } catch (SocketException e) {  
48.             // TODO Auto-generated catch block   
49.             e.printStackTrace();  
50.         } catch (IOException e) {  
51.             // TODO Auto-generated catch block   
52.             e.printStackTrace();  
53.         }  
54.           
55.     }  
56.       
57.       
58.     /** 
59.      * 组播 
60.      * */  
61.     public void multicaseListen(String ipStr, int portNum)  
62.     {  
63.         try {  
64.             InetAddress groupAddress = InetAddress.getByName(ipStr);  
65.             MulticastSocket server = new MulticastSocket(portNum);  
66.               
67.             //加入组播，若地址为非组播地址，抛出IOException   
68.             //若不希望再发送数据到组播地址或者不希望再读取数据，可调用server.leaveGroup(组播地址)   
69.             server.joinGroup(groupAddress);  
70.             MulticastSocket client = new MulticastSocket();  
71.             client.joinGroup(groupAddress);  
72.               
73.             //   
74.               
75.         } catch (UnknownHostException e) {  
76.             // TODO Auto-generated catch block   
77.             e.printStackTrace();  
78.         } catch (IOException e) {  
79.             // 若地址为非组播地址   
80.             e.printStackTrace();  
81.         }  
82.     }  
83.       
84.     public static void main(String[] args) {  
85.         // TODO Auto-generated method stub   
86.   
87.     }  
88.   
89. }  

1. 　　服务器代码:  
2.   
3. 　　import java.net.*;  
4.   
5. 　　import java.nio.*;  
6.   
7. 　　import java.nio.channels.*;  
8.   
9. 　　import java.util.*;  
10.   
11. 　　public class server  
12.   
13. 　　{  
14.   
15. 　　ServerSocketChannel ssc ;  
16.   
17. 　　public void start()  
18.   
19. 　　{  
20.   
21. 　　try  
22.   
23. 　　{  
24.   
25. 　　Selector selector = Selector.open();  
26.   
27. 　　ServerSocketChannel ssc=ServerSocketChannel.open();  
28.   
29. 　　ssc.configureBlocking(false);  
30.   
31. 　　ServerSocket ss=ssc.socket();  
32.   
33. 　　InetSocketAddress address = new InetSocketAddress(55555);  
34.   
35. 　　ss.bind(address);  
36.   
37. 　　ssc.register(selector, SelectionKey.OP_ACCEPT);  
38.   
39. 　　System.out.println("端口注册完毕!");  
40.   
41. 　　while(true)  
42.   
43. 　　{  
44.   
45. 　　selector.select();  
46.   
47. 　　Set<SelectionKey> selectionKeys=selector.selectedKeys();  
48.   
49. 　　Iterator<SelectionKey> iter=selectionKeys.iterator();  
50.   
51. 　　ByteBuffer echoBuffer=ByteBuffer.allocate(20);  
52.   
53. 　　SocketChannel sc;  
54.   
55. 　　while(iter.hasNext())  
56.   
57. 　　{  
58.   
59. 　　SelectionKey key=iter.next();  
60.   
61. 　　if((key.readyOps()&SelectionKey.OP_ACCEPT)==SelectionKey.OP_ACCEPT)  
62.   
63. 　　{  
64.   
65. 　　ServerSocketChannel subssc=(ServerSocketChannel)key.channel();  
66.   
67. 　　sc=subssc.accept();  
68.   
69. 　　sc.configureBlocking(false);  
70.   
71. 　　sc.register(selector, SelectionKey.OP_READ);  
72.   
73. 　　iter.remove();  
74.   
75. 　　System.out.println("有新连接:"+sc);  
76.   
77. 　　}  
78.   
79. 　　else if((key.readyOps()&SelectionKey.OP_READ)==SelectionKey.OP_READ)  
80.   
81. 　　{  
82.   
83. 　　sc=(SocketChannel) key.channel();  
84.   
85. 　　while(true)  
86.   
87. 　　{  
88.   
89. 　　echoBuffer.clear();  
90.   
91. 　　int a;  
92.   
93. 　　try  
94.   
95. 　　{  
96.   
97. 　　a=sc.read(echoBuffer);  
98.   
99. 　　}  
100.   
101. 　　catch(Exception e)  
102.   
103. 　　{  
104.   
105. 　　e.printStackTrace();  
106.   
107. 　　break;  
108.   
109. 　　}  
110.   
111. 　　if(a==-1) break;  
112.   
113. 　　if(a>0)  
114.   
115. 　　{  
116.   
117. 　　byte[] b=echoBuffer.array();  
118.   
119. 　　System.out.println("接收数据: "+new String(b));  
120.   
121. 　　echoBuffer.flip();  
122.   
123. 　　sc.write(echoBuffer);  
124.   
125. 　　System.out.println("返回数据: "+new String(b));  
126.   
127. 　　}  
128.   
129. 　　}  
130.   
131. 　　sc.close();  
132.   
133. 　　System.out.println("连接结束");  
134.   
135. 　　System.out.println("=============================");  
136.   
137. 　　iter.remove();  
138.   
139. 　　}  
140.   
141. 　　}  
142.   
143. 　　}  
144.   
145. 　　}  
146.   
147. 　　catch (Exception e)  
148.   
149. 　　{  
150.   
151. 　　e.printStackTrace();  
152.   
153. 　　}  
154.   
155. 　　}  
156.   
157. 　　}  
158.   
159. 　　客户端代码:  
160.   
161. 　　import java.net.*;  
162.   
163. 　　import java.nio.*;  
164.   
165. 　　import java.nio.channels.*;  
166.   
167. 　　public class client  
168.   
169. 　　{  
170.   
171. 　　public void start()  
172.   
173. 　　{  
174.   
175. 　　try  
176.   
177. 　　{  
178.   
179. 　　SocketAddress address = new InetSocketAddress("localhost",55555);  
180.   
181. 　　SocketChannel client=SocketChannel.open(address);  
182.   
183. 　　client.configureBlocking(false);  
184.   
185. 　　String a="asdasdasdasddffasfas";  
186.   
187. 　　ByteBuffer buffer=ByteBuffer.allocate(20);  
188.   
189. 　　buffer.put(a.getBytes());  
190.   
191. 　　buffer.clear();  
192.   
193. 　　int d=client.write(buffer);  
194.   
195. 　　System.out.println("发送数据: "+new String(buffer.array()));  
196.   
197. 　　while(true)  
198.   
199. 　　{  
200.   
201. 　　buffer.flip();  
202.   
203. 　　int i=client.read(buffer);  
204.   
205. 　　if(i>0)  
206.   
207. 　　{  
208.   
209. 　　byte[] b=buffer.array();  
210.   
211. 　　System.out.println("接收数据: "+new String(b));  
212.   
213. 　　client.close();  
214.   
215. 　　System.out.println("连接关闭!");  
216.   
217. 　　break;  
218.   
219. 　　}  
220.   
221. 　　}  
222.   
223. 　　}  
224.   
225. 　　catch(Exception e)  
226.   
227. 　　{  
228.   
229. 　　e.printStackTrace();  
230.   
231. 　　}  
232.   
233. 　　}  
234.   
235. 　　}  

这几天需要实现一个底层基于UDP的协议，该协议底层使用UDP传输但是具有拥塞控制、超时重发、数据确认等功能又比TCP简单 （RUDP，Reliable UDP）。在实现协议底层的UDP服务时准备使用Java的NIO，在网上查资料都是以TCP为例讲的，于是自己研究了一下基于UDP的NIO。

NIO的思路是基于多路选择的，即由原来的每个连接都由一个线程来等待消息，改为每个连接都在选择器上注册，由选择器来等待。当然NIO引入了很多新的概念，如Channel，Buffer、Charset、Selector等，使得编程更简洁、更面向对象化。

下面贴出用NIO API改造成UDP示例代码，注意其中使用Charset来编码解码的过程（当然Charset还支持很多其他编码不仅局限于默认编码）以及Buffer的使用。

1. package sinpo.usagedemo;   
2.   
3. import java.net.DatagramSocket;   
4. import java.net.InetSocketAddress;   
5. import java.net.SocketAddress;   
6. import java.nio.ByteBuffer;   
7. import java.nio.CharBuffer;   
8. import java.nio.channels.DatagramChannel;   
9. import java.nio.channels.SelectionKey;   
10. import java.nio.channels.Selector;   
11. import java.nio.charset.Charset;   
12. import java.util.Iterator;   
13. import java.util.Set;   
14.   
15. /** 
16.  * @author 徐辛波(sinpo.xu@hotmail.com) Oct 19, 2008 
17.  */  
18. public class UDPServer extends Thread {  
19.     public void run() {  
20.         Selector selector = null;  
21.         try {  
22.             DatagramChannel channel = DatagramChannel.open();  
23.             DatagramSocket socket = channel.socket();  
24.             channel.configureBlocking(false);  
25.             socket.bind(new InetSocketAddress(5057));  
26.   
27.             selector = Selector.open();  
28.             channel.register(selector, SelectionKey.OP_READ);  
29.         } catch (Exception e) {  
30.             e.printStackTrace();  
31.         }  
32.   
33.         ByteBuffer byteBuffer = ByteBuffer.allocate(65536);  
34.         while (true) {  
35.             try {  
36.                 int eventsCount = selector.select();  
37.                 if (eventsCount > 0) {  
38.                     Set selectedKeys = selector.selectedKeys();  
39.                     Iterator iterator = selectedKeys.iterator();  
40.                     while (iterator.hasNext()) {  
41.                         SelectionKey sk = (SelectionKey) iterator.next();  
42.                         iterator.remove();  
43.                         if (sk.isReadable()) {  
44.                             DatagramChannel datagramChannel = (DatagramChannel) sk  
45.                                     .channel();  
46.                             SocketAddress sa = datagramChannel  
47.                                     .receive(byteBuffer);  
48.                             byteBuffer.flip();  
49.   
50.                             // 测试：通过将收到的ByteBuffer首先通过缺省的编码解码成CharBuffer 再输出   
51.                             CharBuffer charBuffer = Charset.defaultCharset()  
52.                                     .decode(byteBuffer);  
53.                             System.out.println("receive message:"  
54.                                     + charBuffer.toString());  
55.                             byteBuffer.clear();  
56.   
57.                             String echo = "This is the reply message from 服务器。";  
58.                             ByteBuffer buffer = Charset.defaultCharset()  
59.                                     .encode(echo);  
60.                             datagramChannel.write(buffer);  
61.                         }  
62.                     }  
63.                 }  
64.             } catch (Exception e) {  
65.                 e.printStackTrace();  
66.             }  
67.         }  
68.   
69.     }  
70.   
71.     public static void main(String[] args) {  
72.         new UDPServer().start();  
73.     }  
74. }  

1. package sinpo.usagedemo;  
2.   
3. import java.net.InetSocketAddress;  
4. import java.net.SocketAddress;  
5. import java.nio.ByteBuffer;  
6. import java.nio.channels.DatagramChannel;  
7. import java.nio.channels.SelectionKey;  
8. import java.nio.channels.Selector;  
9. import java.nio.charset.Charset;  
10. import java.util.Iterator;  
11. import java.util.Set;  
12.   
13. /** 
14.  * @author 徐辛波(sinpo.xu@hotmail.com) Oct 19, 2008 
15.  */  
16. public class UDPClient extends Thread {  
17.     public void run() {  
18.         DatagramChannel channel = null;  
19.         Selector selector = null;  
20.         try {  
21.             channel = DatagramChannel.open();  
22.             channel.configureBlocking(false);  
23.             SocketAddress sa = new InetSocketAddress("localhost", 5057);  
24.             channel.connect(sa);  
25.         } catch (Exception e) {  
26.             e.printStackTrace();  
27.         }  
28.   
29.         try {  
30.             selector = Selector.open();  
31.             channel.register(selector, SelectionKey.OP_READ);  
32.             channel.write(Charset.defaultCharset().encode("Tell me your time"));  
33.         } catch (Exception e) {  
34.             e.printStackTrace();  
35.         }  
36.   
37.         ByteBuffer byteBuffer = ByteBuffer.allocate(100);  
38.         while (true) {  
39.             try {  
40.                 int eventsCount = selector.select();  
41.                 if (eventsCount > 0) {  
42.                     Set selectedKeys = selector.selectedKeys();  
43.                     Iterator iterator = selectedKeys.iterator();  
44.                     while (iterator.hasNext()) {  
45.                         SelectionKey sk = (SelectionKey) iterator.next();  
46.                         iterator.remove();  
47.                         if (sk.isReadable()) {  
48.                             DatagramChannel datagramChannel = (DatagramChannel) sk  
49.                                     .channel();  
50.                             datagramChannel.read(byteBuffer);  
51.                             byteBuffer.flip();  
52.   
53.                             // TODO 将报文转化为RUDP消息并调用RUDP协议处理器来处理   
54.   
55.                             System.out.println(Charset.defaultCharset()  
56.                                     .decode(byteBuffer).toString());  
57.                             byteBuffer.clear();  
58.                             datagramChannel.write(Charset.defaultCharset()  
59.                                     .encode("Tell me your time"));  
60.                         }  
61.                     }  
62.                 }  
63.             } catch (Exception e) {  
64.                 e.printStackTrace();  
65.             }  
66.         }  
67.   
68.     }  
69. }  

import java.io.IOException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.nio.ByteBuffer;
 import java.nio.CharBuffer;
 import java.nio.channels.SelectableChannel;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.ServerSocketChannel;
 import java.nio.channels.SocketChannel;
 import java.nio.charset.CharacterCodingException;
 import java.nio.charset.Charset;
 import java.nio.charset.CharsetDecoder;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Set;
 
 public class NBlockingServer {
 	int port = 8000;
 	int BUFFERSIZE = 1024;
 	Selector selector = null;
 	ServerSocketChannel serverChannel = null;
 	HashMap clientChannelMap = null;// 用来存放每一个客户连接对应的套接字和通道
 
 	public NBlockingServer(int port) {
 		this.clientChannelMap = new HashMap();
 		this.port = port;
 	}
 
 	public void initialize() throws IOException {
 		// 初始化，分别实例化一个选择器，一个服务器端可选择通道
 		this.selector = Selector.open();
 		this.serverChannel = ServerSocketChannel.open();
 		this.serverChannel.configureBlocking(false);
 		InetAddress localhost = InetAddress.getLocalHost();
 		InetSocketAddress isa = new InetSocketAddress(localhost, this.port);
 		this.serverChannel.socket().bind(isa);// 将该套接字绑定到服务器某一可用端口
 	}
 
 	// 结束时释放资源
 	public void finalize() throws IOException {
 		this.serverChannel.close();
 		this.selector.close();
 	}
 
 	// 将读入字节缓冲的信息解码
 	public String decode(ByteBuffer byteBuffer) throws CharacterCodingException {
 		Charset charset = Charset.forName("ISO-8859-1");
 		CharsetDecoder decoder = charset.newDecoder();
 		CharBuffer charBuffer = decoder.decode(byteBuffer);
 		String result = charBuffer.toString();
 		return result;
 	}
 
 	// 监听端口，当通道准备好时进行相应操作
 	public void portListening() throws IOException, InterruptedException {
 		// 服务器端通道注册OP_ACCEPT事件
 		SelectionKey acceptKey = this.serverChannel.register(this.selector,
 				SelectionKey.OP_ACCEPT);
 		// 当有已注册的事件发生时,select()返回值将大于0
 		while (acceptKey.selector().select() > 0) {
 			System.out.println("event happened");
 			// 取得所有已经准备好的所有选择键
 			Set readyKeys = this.selector.selectedKeys();
 			// 使用迭代器对选择键进行轮询
 			Iterator iter = readyKeys.iterator();
 			while (iter.hasNext()) {
 				SelectionKey key = (SelectionKey) iter.next();
 				iter.remove();
 				if (key.isReadable()) {// 如果是通道读准备好事件
 					System.out.println("Readable");
 					// 取得选择键对应的通道和套接字
 					SelectableChannel nextReady = (SelectableChannel) key
 							.channel();
 					Socket socket = (Socket) key.attachment();
 					// 处理该事件，处理方法已封装在类ClientChInstance中
 					this.readFromChannel(socket.getChannel(),
 							(ClientChInstance) this.clientChannelMap
 									.get(socket));
 				} else if (key.isWritable()) {// 如果是通道写准备好事件
 					System.out.println("writeable");
 					// 取得套接字后处理，方法同上
 					Socket socket = (Socket) key.attachment();
 					SocketChannel channel = (SocketChannel) socket.getChannel();
 					this.writeToChannel(channel, "This is from server!");
 				}
 			}
 		}
 	}
 
 	// 对通道的写操作
 	public void writeToChannel(SocketChannel channel, String message)
 			throws IOException {
 		ByteBuffer buf = ByteBuffer.wrap(message.getBytes());
 		int nbytes = channel.write(buf);
 	}
 
 	// 对通道的读操作
 	public void readFromChannel(SocketChannel channel,
 			ClientChInstance clientInstance) throws IOException,
 			InterruptedException {
 		ByteBuffer byteBuffer = ByteBuffer.allocate(BUFFERSIZE);
 		int nbytes = channel.read(byteBuffer);
 		byteBuffer.flip();
 		String result = this.decode(byteBuffer);
 		// 当客户端发出”@exit”退出命令时，关闭其通道
 		if (result.indexOf("@exit") >= 0) {
 			channel.close();
 		} else {
 			clientInstance.append(result.toString());
 			// 读入一行完毕，执行相应操作
 			if (result.indexOf("/n") >= 0) {
 				System.out.println("client input" + result);
 				clientInstance.execute();
 			}
 		}
 	}
 
 	// 该类封装了怎样对客户端的通道进行操作，具体实现可以通过重载execute()方法
 	public class ClientChInstance {
 		SocketChannel channel;
 		StringBuffer buffer = new StringBuffer();
 
 		public ClientChInstance(SocketChannel channel) {
 			this.channel = channel;
 		}
 
 		public void execute() throws IOException {
 			String message = "This is response after reading from channel!";
 			writeToChannel(this.channel, message);
 			buffer = new StringBuffer();
 		}
 
 		// 当一行没有结束时，将当前字窜置于缓冲尾
 		public void append(String values) {
 			buffer.append(values);
 		}
 	}
 
 	// 主程序
 	public static void main(String[] args) {
 		NBlockingServer nbServer = new NBlockingServer(8000);
 		try {
 			nbServer.initialize();
 		} catch (Exception e) {
 			e.printStackTrace();
 			System.exit(-1);
 		}
 		try {
 			nbServer.portListening();
 		} catch (Exception e) {
 			e.printStackTrace();
 		}
 	}
 }

java 异步socket（zz）

http://blog.csdn.net/w01f_cooleagle/article/details/4845244

传统的Socket是阻塞，像ServerSocket在调用accept方法后便处于阻塞状态等待Client端的连接，所以一般会在Server端使用许多线程，对每一个Socket连接分配一个线程。充分利用并发特性来提高性能。 
但这样会带来许多问题： 
１、Server端创建了许多线程来处理Socket连接，而这些线程大部分的时间都在等待连接， 
也就是说这些线程占了资源，真正做事情的时间却不多。也就是说资源利用率比较低，这就会 
直接导致一个问题，可伸缩性比较差，当接受1000个连接还可以，但增加到10000个或更多时性能会很快下降。像Web服务器Jetty和Tomcat6.x都是用了异步通信模式，来接受客户端的连接，从而很大程度上提高了系统的伸缩性，提高了性能。 
2、由于使用多线程，就会使问题变得复杂，事实如果你敢说精通并发编程，说明你太乐观了，并发编程很容易出错，并且你很难发现问题。并且需要互斥访问一些资源，这往往是个瓶颈，会降低并发性。 
异步的Socket可以解决上面的问题,异步的Socket它是非阻塞的,它尝试去连接,但不管是否能够立即建立连接,它都会立即返回,返回之后它便可以做其他的事情了,但连接真正建立成功,就会有相应的事件来通知,这时候你去做连接成功之后的读写操作了.这个过程,整个线程都是处于忙碌状态,所以只需要单个或者很少几个线程,就可以达到阻塞方式的成百上千的线程的性能. 
类似异步Socket功能应运而生，但Java在jdk1.4才引入这个功能，考虑以前Socket的已提供的功能，而且接口很难一致，Java并没有单独设计异步Socket，而是在java nio中引入了SocketChannel之类的通道来处理这个问题，我们会发现这些通道和对应的Socket提供的接口有很大的相似之处，但这些Channel不是关于Socket的抽象，它们并不处理TCP/UDP协议，而是委托给对Socket来处理。 
这种新的Channel可以在非阻塞的模式下操作。通过注册Selector，使用一种类似观察者模式。其实是Selector不断的轮询连接的端口，我们可以通过Selector的select()方法，这个方法是阻塞的，他会更新就绪操作集的键的数目，并作为返回值返回。我们会通常在判断这个返回值如果不为零，则可能有我们感兴趣的事件发生，然后我们可以通过 

来得到键的迭代器，这样我们就可以通过遍历这个集合来判断有没有我们感兴趣的事情发生，如果有我们就做一些处理，没有我们可以把这个键remove掉： 

一、下面我们介绍一下与Socket相关的各种Channel 
与几种Socket对应的提供了一下几种Channel: 
ServerSocketChannel,SocketChannel,DatagramChannel. 
1、ServerSocketChannel: 

我们发现这个Channel根本不支持读写操作，叫Channel有点名不副实了，但Java中Channel本身的抽象也不一定支持读写操作的，需要实现WriteableChannel和ReadableChannnel接口才能支持。其实ServerSocketChannel本身也不是用于读写 
操作的，它通常通过socket() 方法获取相关的ServerSocket,然后通过ServerSocket 方法bind来绑定端口。ServerSocketChannel提供了open()静态工厂方法来创建ServerSocketChannel对象。同时ServerSocketChannel从AbstractSelectableChannel 
继承了： 

这两个方法是最常用的了。通过configureBlocking来设置通道的是否为阻塞模式， 
通过register向给定的选择器注册此通道。 
从上面的过程我们用代码总结一下一般的流程： 

2、SocketChannel： 
SocketChannel通常作为客户端，建立一个对Server端的连接，任何一个SocketChannel 
都是和对应的Socket关联的，但一个Socket并不一定有SocketChannel可以获得。 
同样的SocketChannel也适用静态工厂方法open()来实例化SocketChannel. 
下面来看看最常用的操作： 

我们可以通过 

来建立连接，这个过程是异步的，他会立即返回，如果立即建立连接成功则返回true,否则 
返回false.随后可以通过finishConnect来完成连接的建立。 
另外可通过： 

建立连接 
等价于： 

下面我们演示一下整个的使用过程： 

3、DatagramChannel 
这个Channel与DatagramSocket相对应，提供了基于UDP协议的数据包的套接字的通道。 
UDP协议是无连接的,DatagramChannelt既可以作为Server端,也可以作为Client端,如果想新创建一个DatagramChannel作为Server端来监听，那么需要绑定到特定的端口或地址和端口的组合，一般过程如下: 

但是一个没有绑定特定端口的DatagramChannel仍然是可以接收数据的，事实上会有一个 
动态生成的端口分配给他。不管DatagramChannel是否绑定到了一个端口，任何一个包的发送都会包含它的地址，下面是DatagramChannel提供的发送和接收数据的方法： 

DatagramChannel并不能保证数据能够发送到目的端，因为UDP协议本身就是不可靠的。 
另外我们再看看DatagramChannel提供了以下下几个方法： 

从名字看起来很让人迷惑，因为DatagramChannel就是基于无连接的，为什么还会有 
connect之类的方法呢？ 
其实这个Connect的语义和基于流的Socket是不一样的，这里的连接只是制定了远程的 
地址，这样就可以忽略其他地址发来的数据包了。一但使用完connect方法，就不能像其 
他的地址send数据了。但这里的connect和SockectChannel的connect不同，它是可以 
随时disconnect，然后去connect其他的地址的。但使用了connect方法之后，就可以 
像FileChannel那样read,write数据，而不用指名地址了。