Java I/O（4）：AIO和NIO中的Selector

技术分享 3年前 (2022-10-21) 0 999+

您好，我是湘王，这是我的博客园，欢迎您来，欢迎您再来～

在Java NIO的三大核心中，除了Channel和Buffer，剩下的就是Selector了。有的地方叫它选择器，也有叫多路复用器的（比如Netty）。

之前提过，数据总是从Channel读取到Buffer，或者从Buffer写入到Channel，单个线程可以监听多个Channel——Selector就是这个线程背后的实现机制（所以得名Selector）。

Selector通过控制单个线程处理多个Channel，如果应用打开了多个Channel，但每次传输的流量都很低，使用Selector就会很方便（至于为什么，具体到Netty中再分析）。所以使用Selector的好处就显而易见：用最少的资源实现最多的操作，避免了线程切换带来的开销。

还是以代码为例来演示Selector的作用。新建一个类，在main()方法中输入下面的代码：

/**  * NIO中的Selector  *  * @author xiangwang  */ public class TestSelector {     public static void main(String args[]) throws IOException {         // 创建ServerSocketChannel         ServerSocketChannel channel1 = ServerSocketChannel.open();         channel1.socket().bind(new InetSocketAddress("127.0.0.1", 8080));         channel1.configureBlocking(false);         ServerSocketChannel channel2 = ServerSocketChannel.open();         channel2.socket().bind(new InetSocketAddress("127.0.0.1", 9090));         channel2.configureBlocking(false);          // 创建一个Selector对象         Selector selector = Selector.open();         // 按照字面意思理解，应该是这样的：selector.register(channel, event);         // 但其实是这样的：channel.register(selector, SelectionKey.OP_READ);         // 四种监听事件：         // OP_CONNECT（连接就绪）         // OP_ACCEPT（接收就绪）         // OP_READ（读就绪）         // OP_WRITE（写就绪）         // 注册Channel到Selector，事件一旦被触发，监听随之结束         SelectionKey key1 = channel1.register(selector, SelectionKey.OP_ACCEPT);         SelectionKey key2 = channel2.register(selector, SelectionKey.OP_ACCEPT);          // 模板代码：在编写程序时，大多数时间都是在模板代码中添加相应的业务代码         while(true) {             int readyNum = selector.select();             if (readyNum == 0) {                 continue;             }              Set<SelectionKey> selectedKeys = selector.selectedKeys();             // 轮询             for (SelectionKey key : selectedKeys) {                 Channel channel = key.channel();                 if (key.isConnectable()) {                     if (channel == channel1) {                         System.out.println("channel1连接就绪");                     } else {                         System.out.println("channel2连接就绪");                     }                 } else if (key.isAcceptable()) {                     if (channel == channel1) {                         System.out.println("channel1接收就绪");                     } else {                         System.out.println("channel2接收就绪");                     }                 }                 // 触发后删除，这里不删                 // it.remove();             }         }     } }

代码写好后启动ServerSocketChannel服务，可以看到我这里已经启动成功：

然后在网上下载一个叫做SocketTest.jar的工具（在一些工具网站下载的时候当心中毒，如果不放心，可以私信我，给你地址），双击打开，并按下图方式执行：

点击「Connect」可以看到变化：

然后点击「Disconnect」，再输入「9090」后，再点击「Connect」试试：

可以看到结果显示结果变了：

两次连接，打印了三条信息：说明selector的轮询在起作用（因为Set<SelectionKey>中包含了所有处于监听的SelectionKey）。但是「接收就绪」监听事件仅执行了一次就再不响应。如果感兴趣的话你可以把OP_READ、OP_WRITE这些事件也执行一下试试看。

因为Selector是单线程轮询监听多个Channel，那么如果Selector（线程）之间需要传递数据，怎么办呢？——Pipe登场了。Pipe就是一种用于Selector之间数据传递的「管道」。

先来看个图：

可以清楚地看到它的工作方式。

还是用代码来解释。

/**  * NIO中的Pipe  *  * @author xiangwang  */ public class TestPipe {     public static void main(String args[]) throws IOException {         // 打开管道         Pipe pipe = Pipe.open();          // 将Buffer数据写入到管道         Pipe.SinkChannel sinkChannel = pipe.sink();         ByteBuffer buffer = ByteBuffer.allocate(32);         buffer.put("ByteBuffer".getBytes());         // 切换到写模式         buffer.flip();         sinkChannel.write(buffer);          // 从管道读取数据         Pipe.SourceChannel sourceChannel = pipe.source();         buffer = ByteBuffer.allocate(32);         sourceChannel.read(buffer);         System.out.println(new String(buffer.array()));          // 关闭管道         sinkChannel.close();         sourceChannel.close();     } }

之前说过，同步指的按顺序一次完成一个任务，直到前一个任务完成并有了结果以后，才能再执行后面的任务。而异步指的是前一个任务结束后，并不等待任务结果，而是继续执行后一个任务，在所有任务都「执行」完后，通过任务的回调函数去获得结果。所以异步使得应用性能有了极大的提高。为了更加生动地说明什么是异步，可以来做个实验：

通过调用CompletableFuture.supplyAsync()方法可以很明显地观察到，处于位置2的「这一步先执行」会最先显示，然后才执行位置1的代码。而这就是异步的具体实现。

NIO为了支持异步，升级到了NIO2，也就是AIO。而AIO引入了新的异步Channel的概念，并提供了异步FileChannel和异步SocketChannel的实现。AIO的异步SocketChannel是真正的异步非阻塞I/O。通过代码可以更好地说明：

/**  * AIO客户端  *  * @author xiangwang  */ public class AioClient {     public void start() throws IOException, InterruptedException {         AsynchronousSocketChannel channel = AsynchronousSocketChannel.open();         if (channel.isOpen()) {             // socket接收缓冲区recbuf大小             channel.setOption(StandardSocketOptions.SO_RCVBUF, 128 * 1024);             // socket发送缓冲区recbuf大小             channel.setOption(StandardSocketOptions.SO_SNDBUF, 128 * 1024);             // 保持长连接状态             channel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);             // 连接到服务端             channel.connect(new InetSocketAddress(8080), null,                     new AioClientHandler(channel));             // 阻塞主进程             for(;;) {                 TimeUnit.SECONDS.sleep(1);             }         } else {             throw new RuntimeException("Channel not opened!");         }     }      public static void main(String[] args) throws IOException, InterruptedException {         new AioClient().start();     } }

/**  * AIO客户端CompletionHandler  *  * @author xiangwang  */ public class AioClientHandler implements CompletionHandler<Void, AioClient> {     private final AsynchronousSocketChannel channel;     private final CharsetDecoder decoder = Charset.defaultCharset().newDecoder();     private final BufferedReader input = new BufferedReader(new InputStreamReader(System.in));      public AioClientHandler(AsynchronousSocketChannel channel) {         this.channel = channel;     }     @Override     public void failed(Throwable exc, AioClient attachment) {         throw new RuntimeException("channel not opened!");     }     @Override     public void completed(Void result, AioClient attachment) {         System.out.println("send message to server: ");         try {             // 将输入内容写到buffer             String line = input.readLine();             channel.write(ByteBuffer.wrap(line.getBytes()));             // 在操作系统中的Java本地方法native已经把数据写到了buffer中             // 这里只需要一个缓冲区能接收就行了             ByteBuffer buffer = ByteBuffer.allocate(1024);             while (channel.read(buffer).get() != -1) {                 buffer.flip();                 System.out.println("from server: " + decoder.decode(buffer).toString());                 if (buffer.hasRemaining()) {                     buffer.compact();                 } else {                     buffer.clear();                 }                 // 将输入内容写到buffer                 line = input.readLine();                 channel.write(ByteBuffer.wrap(line.getBytes()));             }         } catch (IOException | InterruptedException | ExecutionException e) {             e.printStackTrace();         }     } }

/**  * AIO服务端  *  * @author xiangwang  */ public class AioServer {     public void start() throws InterruptedException, IOException {         AsynchronousServerSocketChannel channel = AsynchronousServerSocketChannel.open();         if (channel.isOpen()) {             // socket接受缓冲区recbuf大小             channel.setOption(StandardSocketOptions.SO_RCVBUF, 4 * 1024);             // 端口重用，防止进程意外终止，未释放端口，重启时失败             // 因为直接杀进程，没有显式关闭套接字来释放端口，会等待一段时间后才可以重新use这个关口             // 解决办法就是用SO_REUSEADDR             channel.setOption(StandardSocketOptions.SO_REUSEADDR, true);             channel.bind(new InetSocketAddress(8080));         } else {             throw new RuntimeException("channel not opened!");         }         // 处理client连接         channel.accept(null, new AioServerHandler(channel));         System.out.println("server started");         // 阻塞主进程         for(;;) {             TimeUnit.SECONDS.sleep(1);         }     }      public static void main(String[] args) throws IOException, InterruptedException {         AioServer server = new AioServer();         server.start();     } }

/**  * AIO服务端CompletionHandler  *  * @author xiangwang  */ public class AioServerHandler implements CompletionHandler<AsynchronousSocketChannel, Void> {     private final AsynchronousServerSocketChannel serverChannel;     private final CharsetDecoder decoder = Charset.defaultCharset().newDecoder();     private final BufferedReader input = new BufferedReader(new InputStreamReader(System.in));      public AioServerHandler(AsynchronousServerSocketChannel serverChannel) {         this.serverChannel = serverChannel;     }     @Override     public void failed(Throwable exc, Void attachment) {         // 处理下一次的client连接         serverChannel.accept(null, this);     }     @Override     public void completed(AsynchronousSocketChannel result, Void attachment) {         // 处理下一次的client连接，类似链式调用         serverChannel.accept(null, this);         try {             // 将输入内容写到buffer             String line = input.readLine();             result.write(ByteBuffer.wrap(line.getBytes()));             // 在操作系统中的Java本地方法native已经把数据写到了buffer中             // 这里只需要一个缓冲区能接收就行了             ByteBuffer buffer = ByteBuffer.allocate(1024);             while (result.read(buffer).get() != -1) {                 buffer.flip();                 System.out.println("from client: " + decoder.decode(buffer).toString());                 if (buffer.hasRemaining()) {                     buffer.compact();                 } else {                     buffer.clear();                 }                 // 将输入内容写到buffer                 line = input.readLine();                 result.write(ByteBuffer.wrap(line.getBytes()));             }         } catch (InterruptedException | ExecutionException | IOException e) {             e.printStackTrace();         }     } }