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

Posted 2020-07-09 IT·达人

选择器服务器端代码

上一篇文章毫无条理地讲了很多和选择器相关的知识点，下面进入实战，看一下如何写和使用选择器实现服务端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 }

 

代码执行结果

先运行服务端程序：

空白，很正常，因为在监听客户端数据的到来，此时并没有数据。接着运行客户端程序：

看到5个线程的数据已经发送，此时服务端的执行情况是：

数据全部接收到并打印，看到右边的方框还是红色的，说明这5个线程的数据接收、打印完毕之后，再继续等待着客户端的数据的到来。

总结一下Selector的执行两个关键点：

1、注册一个ServerSocketChannel到selector中，这个通道的作用只是为了监听客户端是否有数据到来（这里注意一下有数据到来，意思是假如需要接收100个字节，如果到来了1个字节就算数据到来了），只要有数据到来，就把特定通道注册到selector中，并指定其事件为读事件。

2、ServerSocketChannel和SocketChannel（通道 里面的是客户端的数据）共同存在在Selector中，只要有注册的事件到来，Selector取消阻塞状态，遍历SelectionKey集合，继续注 册读取数据的通道，或者是从通道中读取数据。

 

选择过程的可扩展性

从上面的代码以及之前对于Selector的解读可以看到，Selector可以简化用单线程同时管理多个可选择通道的实现。使用一个线程来为多个通道提供服务，通过消除管理各个线程的额外开销，可能会降低复杂性并可能大幅提升性能。但只使用一个线程来服务所有可选择的通道是不是一个好主意呢？这要看情况。

对单核CPU的系统而言这可能是一个好主意，因为在任何情况下都只有一个线程能够 运行。通过消除在线程之间进行上下文切换带来的额外开销，总吞吐量可以提高。但对于一个多核CPU的系统而言呢？字啊一个有n个CPU的系统上，当一个单 一的线程线性轮流地处理每一个线程时，可能有(n-1)个CPU处于空闲状态。

一种可行的解决办法是使用多个选择器。但是请尽量不要这么做，在大量通道上执行就绪选择并不会有很大的开销，大多数工作是由底层操作系统完成的，管理多个选择器并随机地将通道分派给它们当中的一个并不是这个问题的合理的解决方案。

一种更好的解决方案是对所有的可选择通道使用同一个选择器，并将对就绪选择通道的服务委托给其他线程。开发者只使用一个线程监控通道的就绪状态，至于通道处于就绪状态之后又如何做，有两种可行的做法：

1、使用一个协调好的工作线程池来处理接收到的数据，当然线程池的大小是可以调整的

2、通道根据功能由不同的工作线程来处理，它们可能是日志线程、命令/控制线程、状态请求线程等