Java NIO6：选择器2---代码篇

时间：2016-02-10 22:09:24 阅读：273 评论：0 收藏：0 [点我收藏+]

标签：

选择器服务器端代码

上一篇文章毫无条理地讲了很多和选择器相关的知识点，下面进入实战，看一下如何写和使用选择器实现服务端Socket数据接收的程序，这也是NIO中最核心、最精华的部分。

看一下代码：

 1 public class SelectorServer
 2 {
 3     private static int PORT = 1234;
 4     
 5     public static void main(String[] args) throws Exception
 6     {
 7         // 先确定端口号
 8         int port = PORT;
 9         if (args != null && args.length > 0)
10         {
11             port = Integer.parseInt(args[0]);
12         }
13         // 打开一个ServerSocketChannel
14         ServerSocketChannel ssc = ServerSocketChannel.open();
15         // 获取ServerSocketChannel绑定的Socket
16         ServerSocket ss = ssc.socket();
17         // 设置ServerSocket监听的端口
18         ss.bind(new InetSocketAddress(port));
19         // 设置ServerSocketChannel为非阻塞模式
20         ssc.configureBlocking(false);
21         // 打开一个选择器
22         Selector selector = Selector.open();
23         // 将ServerSocketChannel注册到选择器上去并监听accept事件
24         ssc.register(selector, SelectionKey.OP_ACCEPT);
25         while (true)
26         {
27             // 这里会发生阻塞，等待就绪的通道
28             int n = selector.select();
29             // 没有就绪的通道则什么也不做
30             if (n == 0)
31             {
32                 continue;
33             }
34             // 获取SelectionKeys上已经就绪的通道的集合
35             Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
36             // 遍历每一个Key
37             while (iterator.hasNext())
38             {
39                 SelectionKey sk = iterator.next();
40                 // 通道上是否有可接受的连接
41                 if (sk.isAcceptable())
42                 {
43                     ServerSocketChannel ssc1 = (ServerSocketChannel)sk.channel();
44                     SocketChannel sc = ssc1.accept();
45                     sc.configureBlocking(false);
46                     sc.register(selector, SelectionKey.OP_READ);
47                 }
48                 // 通道上是否有数据可读
49                 else if (sk.isReadable())
50                 {
51                     readDataFromSocket(sk);
52                 }
53                 iterator.remove();
54             }
55         }
56     }
57     
58     private static ByteBuffer bb = ByteBuffer.allocate(1024);
59     
60     // 从通道中读取数据
61     protected static void readDataFromSocket(SelectionKey sk) throws Exception
62     {
63         SocketChannel sc = (SocketChannel)sk.channel();
64         bb.clear();
65         while (sc.read(bb) > 0)
66         {
67             bb.flip();
68             while (bb.hasRemaining())
69             {
70                 System.out.print((char)bb.get());
71             }
72             System.out.println();
73             bb.clear();
74         }
75     }
76 }

代码中已经有了相关的注释，这里继续解释一下：

（1）第8行~第12行，确定要监听的端口号，这里是1234

（2）第13行~第20行，由于选择器管理的是通道（Channel），因此首先要有通道。这里是服务器的程序，因此获取ServerSocketChannel，同时获取它所对应的ServerSocket，设置服务端的Channel为非阻塞模式，并绑定之前确定的端口号1234

（3）第21行~第24行，打开一个选择器，并注册当前通道感兴趣的时间为accept时间，即监听来自客户端的Socket数据

（4）第25行~第28行，调用select()方法等待来自客户端的Socket数据。程序会阻塞在这儿不会往下走，直到客户端有Socket数据的到来为止，所以严格意义上来说，NIO并不是一种非阻塞IO，因为NIO会阻塞在Selector的select()方法上

（5）第29行~第33行，没有什么好说的，如果select()方法获取的数据是0的话，下面的代码都没必要走，当然这是有可能发生的

（6）第34行~第39行，获取到已经就绪的通道的迭代器进行迭代，泛型是选择键SelectionKey，前文讲过，选择键用于封装特定的通道

（7）第40行~第52行，这里是一个关键点、核心点，这里做了两件事情：

　　a）满足isAcceptable()则表示该通道上有数据到来了，此时我们做的事情不是获取该通道->创建一个线程来读取该通道上的数据，这么做就和前面一直讲的阻塞IO没有区别了，也无法发挥出NIO的优势来。我们做的事情只是简单地将对应的SocketChannel注册到选择器上，通过传入OP_READ标记，告诉选择器我们关心新的Socket通道什么时候可以准备好读数据

　　b）满足isReadable()则表示新注册的Socket通道已经可以读取数据了，此时调用readDataFromSocket方法读取SocketChannel中的数据，读取数据的方法前面通道的文章中已经详细讲过了，就不讲了

（8）第53行，将键移除，这一行很重要也是容易忘记的一步操作。加入不remove，将会导致45行中出现空指针异常，原因不难理解，可以自己思考一下。

选择器客户端代码

选择器客户端的代码，没什么要求，只要向服务器端发送数据就可以了。这里选用的是Java NIO4：Socket通道一文中，最后一部分开五个线程向服务端发送数据的程序：

 1 public class SelectorClient
 2 {
 3     private static final String STR = "Hello World!";
 4     private static final String REMOTE_IP = "127.0.0.1";
 5     private static final int THREAD_COUNT = 5;
 6     
 7     private static class NonBlockingSocketThread extends Thread
 8     {
 9         public void run()
10         {
11             try
12             {
13                 int port = 1234;
14                 SocketChannel sc = SocketChannel.open();
15                 sc.configureBlocking(false);
16                 sc.connect(new InetSocketAddress(REMOTE_IP, port));
17                 while (!sc.finishConnect())
18                 {
19                     System.out.println("同" + REMOTE_IP + "的连接正在建立，请稍等！");
20                     Thread.sleep(10);
21                 }
22                 System.out.println("连接已建立，待写入内容至指定ip+端口！时间为" + System.currentTimeMillis());
23                 String writeStr = STR + this.getName();
24                 ByteBuffer bb = ByteBuffer.allocate(writeStr.length());
25                 bb.put(writeStr.getBytes());
26                 bb.flip(); // 写缓冲区的数据之前一定要先反转(flip)
27                 sc.write(bb);
28                 bb.clear();
29                 sc.close();
30             } 
31             catch (IOException e)
32             {
33                 e.printStackTrace();
34             } 
35             catch (InterruptedException e)
36             {
37                 e.printStackTrace();
38             }
39         }
40     }
41     
42     public static void main(String[] args) throws Exception
43     {
44         NonBlockingSocketThread[] nbsts = new NonBlockingSocketThread[THREAD_COUNT];
45         for (int i = 0; i < THREAD_COUNT; i++)
46             nbsts[i] = new NonBlockingSocketThread();
47         for (int i = 0; i < THREAD_COUNT; i++)
48             nbsts[i].start();
49         // 一定要join保证线程代码先于sc.close()运行，否则会有AsynchronousCloseException
50         for (int i = 0; i < THREAD_COUNT; i++)
51             nbsts[i].join();
52     }
53 }