Java IO(1)
Java IO(2)

前边描述的文件IO和Socket IO，也都是阻塞的IO，在执行下一步之前必须等待上一步完成. 否则就得一直等待在原地.

  • 可读事件: 如果socket缓冲区有数据可以读，对应的fd就是readable，应用可以从fd中把数据读取走，进行应用处理。
  • 可写事件: 如果socket缓冲区有空间可以写，对应的fd就是writable，应用可以数据写入到fd缓冲区，交给操作系统写入到协议栈

非阻塞IO

顾名思义，和阻塞IO相对应，在文件IO中的阻塞操作可以通过设置在阻塞中变得非阻塞，完成丑小鸭到丑大鸭的蜕变.

示例

  1. /**
  2.  * @author chenshun00@gmail.com
  3.  * @module parse
  4.  * @since 2020/11/30 10:21 下午
  5.  */
  6. public class TestIO {
  8.     public static void main(String[] args) throws IOException {
  9.         //开启一个ServerSocketChannel，对应了Socket IO中的ServerSocket
  10.         ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
  11.         final Selector selector = Selector.open();
  12.         serverSocketChannel.socket().setReuseAddress(true);
  13.         serverSocketChannel.socket().bind(new InetSocketAddress("localhost", 9998));
  14.         //配置非阻塞,如果设置成true还是阻塞的配置
  15.         serverSocketChannel.configureBlocking(false);
  16.         //注意这里的第三个参数，会attach到SelectorKey上
  17.         serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, serverSocketChannel);
  18.         System.out.println("-- server ready");
  19.         while (true) {
  20.             //jdk 11才有的方法
  21.             selector.select(key -> {
  22.                 if (!key.isValid()) {
  23.                     return;
  24.                 }
  25.                 //获取到attach的key
  26.                 final ServerSocketChannel ch = (ServerSocketChannel) key.attachment();
  27.                 try {
  28.                     if (key.isAcceptable()) {
  29.                         //获取endpoint，对应了serverSocket.accpet()返回的socket
  30.                         SocketChannel client_ch = ch.accept();
  31.                         if (client_ch != null) { // accept() may return null...
  32.                             System.out.printf("accepted connection from %s\n", client_ch.getRemoteAddress());
  33.                             //客户端也需要配置成非阻塞
  34.                             client_ch.configureBlocking(false);
  35.                             //监听可读时间
  36.                             client_ch.register(selector, SelectionKey.OP_READ, key.attachment());
  37.                         }
  38.                     } else if (key.isReadable()) {
  39.                         //整体的流程就是从socket中读取数据
  40.                         //然后写入到队列中，然后注册监听可写事件
  41.                         //当触发可写事件时，从队列中获取数据并写回
  42.                         SocketChannel socketChannel = (SocketChannel) key.channel();
  43.                         socketChannel.register(selector, SelectionKey.OP_WRITE);
  44.                         ByteBuffer buffer = ByteBuffer.allocate(1024);
  45.                         int read = socketChannel.read(buffer);
  46.                         buffer.flip();
  47.                         if (read == -1) {
  48.                             throw new IOException("Socket closed");
  49.                         }
  50.                         String result = new String(buffer.array()).trim();
  51.                         System.out.println("receive:" + result);
  52.                         ByteBuffer writeBuffer = ByteBuffer.wrap(ACK);
  53.                         Queue<Object> pendingWrites = channelToPendingWrites.get(key.channel());
  54.                         if (pendingWrites == null) {
  55.                             synchronized (channelToPendingWrites) {
  56.                                 pendingWrites = channelToPendingWrites.get(key.channel());
  57.                                 if (pendingWrites == null) {
  58.                                     pendingWrites = new ConcurrentLinkedQueue<>();
  59.                                     channelToPendingWrites.put(key.channel(), pendingWrites);
  60.                                 }
  61.                             }
  62.                         }
  63.                         pendingWrites.add(writeBuffer);
  64.                         socketChannel.register(selector, SelectionKey.OP_WRITE);
  65.                     } else if (key.isWritable()) {
  66.                         //参考可读事件的注释
  67.                         SocketChannel socketChannel = (SocketChannel) key.channel();
  68.                         final Queue<Object> objects = channelToPendingWrites.get(socketChannel);
  69.                         if (objects == null || objects.size() == 0) {
  70.                             final ByteBuffer allocate = ByteBuffer.allocate(12);
  71.                             allocate.put("hello,world!".getBytes());
  72.                             socketChannel.write(allocate);
  73.                         } else {
  74.                             final Object poll = objects.poll();
  75.                             socketChannel.write((ByteBuffer) poll);
  76.                         }
  77.                         socketChannel.register(selector, SelectionKey.OP_READ);
  78.                     }
  79.                 } catch (Exception e) {
  80.                     e.printStackTrace();
  81.                 }
  82.             }, 10000L);
  83.         }
  84.     }
  86.     private static final Map<SelectableChannel, Queue<Object>> channelToPendingWrites = new ConcurrentHashMap<>();
  88.     private static final byte[] ACK = "Data logged successfully\n".getBytes();
  89. }