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

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

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

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

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

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

    1. public static void main(String args[]) throws IOException {
    2.     // 创建ServerSocketChannel
    3.     ServerSocketChannel channel1 = ServerSocketChannel.open();
    4.     channel1.socket().bind(new InetSocketAddress("127.0.0.1", 8080));
    5.     channel1.configureBlocking(false);
    6.     ServerSocketChannel channel2 = ServerSocketChannel.open();
    7.     channel2.socket().bind(new InetSocketAddress("127.0.0.1", 9090));
    8.     channel2.configureBlocking(false);
    9.  
    10.     // 创建一个Selector对象
    11.     Selector selector = Selector.open();
    12.     // 按照字面意思理解，应该是这样的：selector.register(channel, event);
    13.     // 但其实是这样的：channel.register(selector, SelectionKey.OP_READ);
    14.     // 四种监听事件：
    15.     // OP_CONNECT（连接就绪）
    16.     // OP_ACCEPT（接收就绪）
    17.     // OP_READ（读就绪）
    18.     // OP_WRITE（写就绪）
    19.     // 注册Channel到Selector，事件一旦被触发，监听随之结束
    20.     SelectionKey key1 = channel1.register(selector, SelectionKey.OP_ACCEPT);
    21.     SelectionKey key2 = channel2.register(selector, SelectionKey.OP_ACCEPT);
    22.  
    23.     // 模板代码：在编写程序时，大多数时间都是在模板代码中添加相应的业务代码
    24.     while(true) {
    25.         int readyNum = selector.select();
    26.         if (readyNum == 0) {
    27.             continue;
    28.         }
    29.  
    30.         Set selectedKeys = selector.selectedKeys();
    31.         // 轮询
    32.         for (SelectionKey key : selectedKeys) {
    33.             Channel channel = key.channel();
    34.             if (key.isConnectable()) {
    35.                 if (channel == channel1) {
    36.                     System.out.println("channel1连接就绪");
    37.                 } else {
    38.                     System.out.println("channel2连接就绪");
    39.                 }
    40.             } else if (key.isAcceptable()) {
    41.                 if (channel == channel1) {
    42.                     System.out.println("channel1接收就绪");
    43.                 } else {
    44.                     System.out.println("channel2接收就绪");
    45.                 }
    46.             }
    47.             // 触发后删除，这里不删
    48.             // it.remove();
    49.         }
    50.     }
    51. }

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

     

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

     

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

     

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

     

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

     

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

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

    先来看个图：

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

    还是用代码来解释。

    1. public static void main(String args[]) throws IOException {
    2.  
    3.     // 打开管道
    4.     Pipe pipe = Pipe.open();
    5.  
    6.     // 将Buffer数据写入到管道
    7.     Pipe.SinkChannel sinkChannel = pipe.sink();
    8.     ByteBuffer buffer = ByteBuffer.allocate(32);
    9.     buffer.put("ByteBuffer".getBytes());
    10.     // 切换到写模式
    11.     buffer.flip();
    12.     sinkChannel.write(buffer);
    13.  
    14.     // 从管道读取数据
    15.     Pipe.SourceChannel sourceChannel = pipe.source();
    16.     buffer = ByteBuffer.allocate(32);
    17.     sourceChannel.read(buffer);
    18.     System.out.println(new String(buffer.array()));
    19.  
    20.     // 关闭管道
    21.     sinkChannel.close();
    22.     sourceChannel.close();
    23.  
    24. }

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

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

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

    1. /**
    2.  * AIO客户端
    3.  *
    4.  * @author xiangwang
    5.  */
    6. public class AioClient {
    7.     public void start() throws IOException, InterruptedException {
    8.         AsynchronousSocketChannel channel = AsynchronousSocketChannel.open();
    9.         if (channel.isOpen()) {
    10.             // socket接收缓冲区recbuf大小
    11.             channel.setOption(StandardSocketOptions.SO_RCVBUF, 128 * 1024);
    12.             // socket发送缓冲区recbuf大小
    13.             channel.setOption(StandardSocketOptions.SO_SNDBUF, 128 * 1024);
    14.             // 保持长连接状态
    15.             channel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
    16.             // 连接到服务端
    17.             channel.connect(new InetSocketAddress(8080), null,
    18.                     new AioClientHandler(channel));
    19.             // 阻塞主进程
    20.             for(;;) {
    21.                 TimeUnit.SECONDS.sleep(1);
    22.             }
    23.         } else {
    24.             throw new RuntimeException("Channel not opened!");
    25.         }
    26.     }
    27.  
    28.     public static void main(String[] args) throws IOException, InterruptedException {
    29.         new AioClient().start();
    30.     }
    31. }

     

    1. /**
    2.  * AIO客户端CompletionHandler
    3.  *
    4.  * @author xiangwang
    5.  */
    6. public class AioClientHandler implements CompletionHandler {
    7.     private final AsynchronousSocketChannel channel;
    8.     private final CharsetDecoder decoder = Charset.defaultCharset().newDecoder();
    9.     private final BufferedReader input = new BufferedReader(new InputStreamReader(System.in));
    10.  
    11.     public AioClientHandler(AsynchronousSocketChannel channel) {
    12.         this.channel = channel;
    13.     }
    14.     @Override
    15.     public void failed(Throwable exc, AioClient attachment) {
    16.         throw new RuntimeException("channel not opened!");
    17.     }
    18.     @Override
    19.     public void completed(Void result, AioClient attachment) {
    20.         System.out.println("send message to server: ");
    21.         try {
    22.             // 将输入内容写到buffer
    23.             String line = input.readLine();
    24.             channel.write(ByteBuffer.wrap(line.getBytes()));
    25.             // 在操作系统中的Java本地方法native已经把数据写到了buffer中
    26.             // 这里只需要一个缓冲区能接收就行了
    27.             ByteBuffer buffer = ByteBuffer.allocate(1024);
    28.             while (channel.read(buffer).get() != -1) {
    29.                 buffer.flip();
    30.                 System.out.println("from server: " + decoder.decode(buffer).toString());
    31.                 if (buffer.hasRemaining()) {
    32.                     buffer.compact();
    33.                 } else {
    34.                     buffer.clear();
    35.                 }
    36.                 // 将输入内容写到buffer
    37.                 line = input.readLine();
    38.                 channel.write(ByteBuffer.wrap(line.getBytes()));
    39.             }
    40.         } catch (IOException | InterruptedException | ExecutionException e) {
    41.             e.printStackTrace();
    42.         }
    43.     }
    44. }

     

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

    1. /**
    2.  * AIO服务端CompletionHandler
    3.  *
    4.  * @author xiangwang
    5.  */
    6. public class AioServerHandler implements CompletionHandler {
    7.     private final AsynchronousServerSocketChannel serverChannel;
    8.     private final CharsetDecoder decoder = Charset.defaultCharset().newDecoder();
    9.     private final BufferedReader input = new BufferedReader(new InputStreamReader(System.in));
    10.  
    11.     public AioServerHandler(AsynchronousServerSocketChannel serverChannel) {
    12.         this.serverChannel = serverChannel;
    13.     }
    14.     @Override
    15.     public void failed(Throwable exc, Void attachment) {
    16.         // 处理下一次的client连接
    17.         serverChannel.accept(null, this);
    18.     }
    19.     @Override
    20.     public void completed(AsynchronousSocketChannel result, Void attachment) {
    21.         // 处理下一次的client连接，类似链式调用
    22.         serverChannel.accept(null, this);
    23.         try {
    24.             // 将输入内容写到buffer
    25.             String line = input.readLine();
    26.             result.write(ByteBuffer.wrap(line.getBytes()));
    27.             // 在操作系统中的Java本地方法native已经把数据写到了buffer中
    28.             // 这里只需要一个缓冲区能接收就行了
    29.             ByteBuffer buffer = ByteBuffer.allocate(1024);
    30.             while (result.read(buffer).get() != -1) {
    31.                 buffer.flip();
    32.                 System.out.println("from client: " + decoder.decode(buffer).toString());
    33.                 if (buffer.hasRemaining()) {
    34.                     buffer.compact();
    35.                 } else {
    36.                     buffer.clear();
    37.                 }
    38.                 // 将输入内容写到buffer
    39.                 line = input.readLine();
    40.                 result.write(ByteBuffer.wrap(line.getBytes()));
    41.             }
    42.         } catch (InterruptedException | ExecutionException | IOException e) {
    43.             e.printStackTrace();
    44.         }
    45.     }
    46. }

    执行测试后显示，不管是在客户端还是在服务端，读写完全是异步的。

    感谢您的大驾光临！咨询技术、产品、运营和管理相关问题，请关注后留言。欢迎骚扰，不胜荣幸～

  • 相关阅读:
    【软件测试笔试题】阿里巴巴（中国）网络技术有限公司
    (完全解决)pycharm运行或者调试项目的时候报错：test setup failed
    3.4 常用操作
    ROS2踩坑记录
    大学生餐饮主题网页制作 美食网页设计模板 学生静态网页作业成品 dreamweaver美食甜品蛋糕HTML网站制作
    DevEco Hvigor高效编译，构建过程新秘籍
    华为云 云证书（SSL）管理服务
    C. Joyboard
    数组元素全排列Java
    BertTokenizer 使用方法
  • 原文地址：https://blog.csdn.net/lostrex/article/details/127436894