• 非阻塞模式下，无法保证把 buffer 中所有数据都写入 channel，因此需要追踪 write 方法的返回值（代表实际写入字节数）
  • 用 selector 监听所有 channel 的可写事件，每个 channel 都需要一个 key 来跟踪 buffer，但这样又会导致占用内存过多，就有两阶段策略
    • 当消息处理器第一次写入消息时，才将 channel 注册到 selector 上
    • selector 检查 channel 上的可写事件，如果所有的数据写完了，就取消 channel 的注册
    • 如果不取消，会每次可写均会触发 write 事件

服务端

  1. package cn.inetty.nio.selector3;
  3. import lombok.extern.slf4j.Slf4j;
  5. import java.io.IOException;
  6. import java.net.InetSocketAddress;
  7. import java.nio.ByteBuffer;
  8. import java.nio.channels.SelectionKey;
  9. import java.nio.channels.Selector;
  10. import java.nio.channels.ServerSocketChannel;
  11. import java.nio.channels.SocketChannel;
  12. import java.nio.charset.Charset;
  13. import java.util.Iterator;
  15. import static cn.inetty.nio.ByteBufferUtil.debugAll;
  17. @Slf4j
  18. public class Server {
  19.     public static void main(String[] args) throws IOException {
  20.         ServerSocketChannel open = ServerSocketChannel.open();
  21.         open.configureBlocking(false);
  23.         Selector selector = Selector.open();
  24.         open.register(selector, SelectionKey.OP_ACCEPT);
  25.         open.bind(new InetSocketAddress(8888));
  27.         while (true) {
  28.             selector.select();
  29.             Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
  30.             int w = 0;
  31.             while (iterator.hasNext()) {
  32.                 SelectionKey key = iterator.next();
  33.                 iterator.remove();
  34.                 if (key.isAcceptable()) {
  35.                     SocketChannel sc = open.accept();
  36.                     sc.configureBlocking(false);
  37.                     SelectionKey selectionKey = sc.register(selector, 0, null);
  39.                     // 向客户端发送大量数据
  40.                     StringBuilder sb = new StringBuilder();
  41.                     for (int i = 0; i < 3000000; i++) {
  42.                         sb.append("a");
  43.                     }
  44.                     ByteBuffer buffer = Charset.defaultCharset().encode(sb.toString());
  45.                     // 返回值 = 实际写入的字节数
  46.                     if (buffer.hasRemaining()) {
  47.                         // 关注可写事件 为可能的避免覆盖原有的兴趣集，在基础上添加
  48.                         selectionKey.interestOps(selectionKey.interestOps() + SelectionKey.OP_WRITE);
  49. //                        selectionKey.interestOps(selectionKey.interestOps() | SelectionKey.OP_WRITE);
  50.                         selectionKey.attach(buffer);
  51.                     }
  52.                 } else if (key.isWritable()) {
  53.                     ByteBuffer buffer = (ByteBuffer)key.attachment();
  54.                     SocketChannel channel = (SocketChannel) key.channel();
  55.                     int write = channel.write(buffer);
  56.                     w+=write;
  57.                     System.out.println(w);
  59.                     // 清除资源
  60.                     if (!buffer.hasRemaining()) {
  61.                         key.attach(null);
  62.                         key.interestOps(key.interestOps() - SelectionKey.OP_WRITE);
  63.                     }
  64.                 }
  65.             }
  66.         }
  67.     }
  68. }

客户端

  1. package cn.inetty.nio.selector3;
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.SocketChannel;
  7. import java.nio.charset.Charset;
  9. public class Client1 {
  10.     public static void main(String[] args) throws IOException {
  12.         SocketChannel socketChannel = SocketChannel.open();
  13.         socketChannel.connect(new InetSocketAddress("localhost", 8888));
  14.         int count = 0;
  15.         while (true) {
  16.             ByteBuffer buffer = ByteBuffer.allocate(1024*1024);
  17.             count += socketChannel.read(buffer);
  18.             System.out.println(count);
  19.             buffer.clear();
  20.         }
  21.     }
  22. }