2.3.5  NIO创建的TimeClient源码分析（1）

我们首先还是看下如何对TimeClient进行改造。

代码清单2-9  NIO时间服务器客户端  TimeClient
 

16.     if (args != null && args.length > 0) {  
17.         try {  
18.         port = Integer.valueOf(args[0]);  
19.         } catch (NumberFormatException e) {  
20.         // 采用默认值  
21.         }  
22.     }  
23.     new Thread(new TimeClientHandle("127.0.0.1", port), "TimeClient- 001")  
24.         .start();  
25.     }  
26. } 

与之前唯一不同的地方在于通过创建TimeClientHandle线程来处理异步连接和读写操作，由于TimeClient非常简单且变更不大，这里重点分析TimeClientHandle，代码如下。

代码清单2-10  NIO时间服务器客户端  TimeClientHandle
 

1.  package com.phei.netty.nio;  
2.  import java.io.IOException;  
3.  import java.net.InetSocketAddress;  
4.  import java.nio.ByteBuffer;  
5.  import java.nio.channels.SelectionKey;  
6.  import java.nio.channels.Selector;  
7.  import java.nio.channels.SocketChannel;  
8.  import java.util.Iterator;  
9.  import java.util.Set;  
10.   
11. /**  
12.  * @author Administrator  
13.  * @date 2014年2月16日  
14.  * @version 1.0  
15.  */  
16. public class TimeClientHandle implements Runnable {  
17.     private String host;  
18.     private int port;  
19.     private Selector selector;  
20.     private SocketChannel socketChannel;  
21.     private volatile boolean stop;  
22.   
23.     public TimeClientHandle(String host, int port) {  
24.     this.host = host == null   "127.0.0.1" : host;  
25.     this.port = port;  
26.     try {  
27.         selector = Selector.open();  
28.         socketChannel = SocketChannel.open();  
29.         socketChannel.configureBlocking(false);  
30.     } catch (IOException e) {  
31.         e.printStackTrace();  
32.         System.exit(1);  
33.     }  
34.     }  
35.   
36.     /*  
37.      * (non-Javadoc)  
38.      *   
39.      * @see java.lang.Runnable#run()  
40.      */  
41.     @Override  
42.     public void run() {  
43.     try {  
44.         doConnect();  
45.     } catch (IOException e) {  
46.         e.printStackTrace();  
47.         System.exit(1);  
48.     }  
49.     while (!stop) {  
50.         try {  
51.         selector.select(1000);  
52.         Set<SelectionKey> selectedKeys = selector.selectedKeys();  
53.         Iterator<SelectionKey> it = selectedKeys.iterator();  
54.         SelectionKey key = null;  
55.         while (it.hasNext()) {  
56.             key = it.next();  
57.             it.remove();  
58.             try {  
59.             handleInput(key);  
60.             } catch (Exception e) {  
61.             if (key != null) {  
62.                 key.cancel();  
63.                 if (key.channel() != null)  
64.                 key.channel().close();  
65.             }  
66.             }  
67.         }  
68.         } catch (Exception e) {  
69.         e.printStackTrace();  
70.         System.exit(1);  
71.         }  
72.     }  
73.   
74.     // 多路复用器关闭后，所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭，所以不需要重复释放资源  
75.     if (selector != null)  
76.         try {  
77.         selector.close();  
78.         } catch (IOException e) {  
79.         e.printStackTrace();  
80.         }  
81.     }  
82.   
83.     private void handleInput(SelectionKey key) throws IOException {  
84.   
85.     if (key.isValid()) {  
86.         // 判断是否连接成功  
87.         SocketChannel sc = (SocketChannel) key.channel();  
88.         if (key.isConnectable()) {  
89.         if (sc.finishConnect()) {  
90.             sc.register(selector, SelectionKey.OP_READ);  
91.             doWrite(sc);  
92.         } else  
93.             System.exit(1);// 连接失败，进程退出  
94.         }  
95.         if (key.isReadable()) {  
96.         ByteBuffer readBuffer = ByteBuffer.allocate(1024);  
97.         int readBytes = sc.read(readBuffer);  
98.         if (readBytes > 0) {  
99.             readBuffer.flip();  
100.                byte[] bytes = new byte[readBuffer.remaining()];  
101.                readBuffer.get(bytes);  
102.                String body = new String(bytes, "UTF-8");  
103.                System.out.println("Now is : " + body);  
104.                this.stop = true;  
105.            } else if (readBytes < 0) {  
106.                // 对端链路关闭  
107.                key.cancel();  
108.                sc.close();  
109.            } else  
110.                ; // 读到0字节，忽略  
111.            }  
112.        }  
113.      
114.        }  
115.      
116.        private void doConnect() throws IOException {  
117.        // 如果直接连接成功，则注册到多路复用器上，发送请求消息，读应答  
118.        if(socketChannel.connect(new InetSocketAddress(host, port))) {  
119.            socketChannel.register(selector, SelectionKey.OP_READ);  
120.            doWrite(socketChannel);  
121.        } else  
122.            socketChannel.register(selector, SelectionKey.OP_CONNECT);  
123.        }  
124.      
125.        private void doWrite(SocketChannel sc) throws IOException {  
126.        byte[] req = "QUERY TIME ORDER".getBytes();  
127.        ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);  
128.        writeBuffer.put(req);  
129.        writeBuffer.flip();  
130.        sc.write(writeBuffer);  
131.        if (!writeBuffer.hasRemaining())  
132.            System.out.println("Send order 2 server succeed.");  
133.        }  
134.    } 

喜欢的朋友可以添加我们的微信账号：

51CTO读书频道二维码

51CTO读书频道活动讨论群：342347198