在 server 端收到请求请求并且进入 unsafe.read 后，开始对channel进行处理，这里主要描述对 OP_ACCEEPT 的处理. 

  1.         ![image.png](https://cdn.nlark.com/yuque/0/2020/png/191132/1608991963596-e9d78cb0-9093-41bf-9012-09578a3b159d.png#align=left&display=inline&height=541&margin=%5Bobject%20Object%5D&name=image.png&originHeight=1082&originWidth=1078&size=74441&status=done&style=none&width=539)<br />这里的 `unsafe` 还是 `NioServerSocketChannel` 中引用的 `unsafe` . <br />`unsafe` 对应关系
  • NioServerSocketChannel ==> NioMessageUnsafe
  • NioSocketChannel ==> NioSocketChannelUnsafe(NioByteUnsafe)

说明，服务端启动时，accept到的数据是在NioMessageUnsafe中进行处理的，而对应的客户端channel则是由NioSocketChannelUnsafe(NioByteUnsafe) 来进行读取处理.

读取细节

  1. private final class NioMessageUnsafe extends AbstractNioUnsafe {
  3.         private final List<Object> readBuf = new ArrayList<Object>();
  5.         @Override
  6.         public void read() {
  7.             //确保是在同一个线程进行处理
  8.             assert eventLoop().inEventLoop();
  9.             //获取NioServerSocketChannel启动的配置，配置也很简单
  10.             final ChannelConfig config = AbstractNioMessageChannel.this.config();
  11.             //获取NioServerSocketChannel对应的pipeline
  12.             final ChannelPipeline pipeline = AbstractNioMessageChannel.this.pipeline();
  13.             //一个配置信息(动态调节buffer的大小)
  14.             final RecvByteBufAllocator.Handle allocHandle =
  15.                 AbstractNioMessageChannel.this.unsafe().recvBufAllocHandle();
  16.             //跳过
  17.             allocHandle.reset(config);
  19.             boolean closed = false;
  20.             Throwable exception = null;
  21.             try {
  22.                 try {
  23.                     do {
  24.                         //这里建立OP_ACCEPT阶段，所以实现比较简单
  25.                         int localRead = doReadMessages(readBuf);
  26.                         if (localRead == 0) {
  27.                             break;
  28.                         }
  29.                         if (localRead < 0) {
  30.                             closed = true;
  31.                             break;
  32.                         }
  34.                         allocHandle.incMessagesRead(localRead);
  35.                     } while (allocHandle.continueReading());
  36.                 } catch (Throwable t) {
  37.                     exception = t;
  38.                 }
  40.                 int size = readBuf.size();
  41.                 for (int i = 0; i < size; i ++) {
  42.                     readPending = false;
  43.                     //对于每一个NioSocketChannel 都触发一下read事件.. 经由pipeline传播该事件
  44.                     pipeline.fireChannelRead(readBuf.get(i));
  45.                 }
  46.                 readBuf.clear();
  47.                 //调节buffer是增是减
  48.                 allocHandle.readComplete();
  49.                 //触发读取完成事件
  50.                 pipeline.fireChannelReadComplete();
  52.                 if (exception != null) {
  53.                     closed = closeOnReadError(exception);
  54.                     //触发异常事件
  55.                     pipeline.fireExceptionCaught(exception);
  56.                 }
  58.                 if (closed) {
  59.                     inputShutdown = true;
  60.                     if (isOpen()) {
  61.                         close(voidPromise());
  62.                     }
  63.                 }
  64.             } finally {
  65.                 // Check if there is a readPending which was not processed yet.
  66.                 // This could be for two reasons:
  67.                 // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
  68.                 // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
  69.                 //
  70.                 // See https://github.com/netty/netty/issues/2254
  71.                 if (!readPending && !config.isAutoRead()) {
  72.                     removeReadOp();
  73.                 }
  74.             }
  75.         }
  76.     }
  • fireChannelRead:每一次读取都会触发一次
  • fireChannelReadComplete: 最终完成的时候读取一次

如果一次TCP传输不能够传输完成，那么fireChannelRead中的数据是不能直接作用于业务的. 因为数据不全.
可以在fireChannelReadComplete进行处理.

doReadMessages

初始化一个NioSocketChannel 用来代表 服务端对端的socket (即客户端socket) 

  1.     @Override
  2.     protected int doReadMessages(List<Object> buf) throws Exception {
  3.         SocketChannel ch = SocketUtils.accept(javaChannel());
  5.         try {
  6.             if (ch != null) {
  7.                 buf.add(new NioSocketChannel(this, ch));
  8.                 return 1;
  9.             }
  10.         } catch (Throwable t) {
  11.             logger.warn("Failed to create a new channel from an accepted socket.", t);
  13.             try {
  14.                 ch.close();
  15.             } catch (Throwable t2) {
  16.                 logger.warn("Failed to close a socket.", t2);
  17.             }
  18.         }
  20.         return 0;
  21.     }

pipeline 传播读取事件

  1. pipeline.fireChannelRead(readBuf.get(i));

同理是从 head —-> tail …
那么现在pipeline上除去head和tail之外还有有多少个Context呢. 

  1.    @Override
  2.     public void bind(Integer port) {
  3.         boss = new NioEventLoopGroup(1);
  4.         work = new NioEventLoopGroup(1);
  6.         try {
  7.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  8.             final ServerTransactionHandler serverTransactionHandler = new ServerTransactionHandler(this.applicationContext);
  9.             serverBootstrap.group(boss, work)
  10.                     .channel(NioServerSocketChannel.class)
  11.                     .handler(new LoggingHandler())
  12.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  13.                         @Override
  14.                         public void initChannel(SocketChannel ch) {
  15.                             ch.pipeline()
  16.                                     .addLast(new MessageDecoder(serialization))
  17.                                     .addLast(new MessageEncoder(serialization))
  18.                                     .addLast(new IdleStateHandler(20, 40, 60))
  19.                                     .addLast(serverTransactionHandler);
  20.                         }
  21.                     }).option(ChannelOption.SO_BACKLOG, 128)
  22.                     .childOption(ChannelOption.SO_KEEPALIVE, true);
  23.             channelFuture = serverBootstrap.bind(port).sync();
  24.         } catch (Exception e) {
  25.             e.printStackTrace();
  26.         }
  27.     }

image.png
ServerBootstrapAcceptor: 客户端链接的channel进来时通过这里的 read 配置 childOption ,随后使用 work 线程组将客户端的channel绑定到 work 的一个线程上，这里的过程和服务端的注册过程一致，少有的几个区别分别是

  • 线程组由boss变成了work(意味着绑定的线程由work替换了boss)
  • channel由server变成了clinet
  • unsafe由NioMessageUnsafe变成了NioSocketChannelUnsafe
  • pipeline的实例话过程依旧，但是流水线的节点变成了配置的以下4个.
    • MessageDecoder
    • MessageEncoder
    • IdleStateHandler
    • serverTransactionHandler

核心关注的莫过于 initChannel 方法在何时被调用.
服务启动

pipeline 传播读取完成事件

到这一步其实没有什么区别了. 只不过触发的事件不一致而已. 可能出现的错误也在上边进行了描述.