所以服务端Channel的Pipeline在传播ChannelRead事件时首先会从HeadContext处理器开始，然后传播到ServerBootstrapAcceptor处理器，最后传播到TailContext处理器结束。

 

(3)服务端Channel默认的Pipeline处理器

首先，服务端启动时会给服务端Channel的Pipeline添加一个ServerBootstrapAcceptor处理器。

//Bootstrap sub-class which allows easy bootstrap of ServerChannel public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {     ...     @Override     void init(Channel channel) throws Exception {         //1.设置服务端Channel的Option与Attr         final Map<ChannelOption<?>, Object> options = options0();         synchronized (options) {             channel.config().setOptions(options);         }         final Map<AttributeKey<?>, Object> attrs = attrs0();         synchronized (attrs) {             for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {                 @SuppressWarnings("unchecked")                 AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();                 channel.attr(key).set(e.getValue());             }         }                 //2.设置客户端Channel的Option与Attr         final EventLoopGroup currentChildGroup = childGroup;         final ChannelHandler currentChildHandler = childHandler;         final Entry<ChannelOption<?>, Object>[] currentChildOptions;         final Entry<AttributeKey<?>, Object>[] currentChildAttrs;         synchronized (childOptions) {             currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));         }         synchronized (childAttrs) {             currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));         }                 //3.配置服务端启动逻辑         ChannelPipeline p = channel.pipeline();         //p.addLast()用于定义服务端启动过程中需要执行哪些逻辑         p.addLast(new ChannelInitializer<Channel>() {             @Override             public void initChannel(Channel ch) throws Exception {                 //一.添加用户自定义的Handler，注意这是handler，而不是childHandler                 final ChannelPipeline pipeline = ch.pipeline();                 ChannelHandler handler = config.handler();                 if (handler != null) pipeline.addLast(handler);                 //二.添加一个特殊的Handler用于接收新连接                 //自定义的childHandler会作为参数传入连接器ServerBootstrapAcceptor                 ch.eventLoop().execute(new Runnable() {                     @Override                     public void run() {                         //调用DefaultChannelPipeline的addLast()方法                         pipeline.addLast(new ServerBootstrapAcceptor(                             currentChildGroup,                              currentChildHandler,                              currentChildOptions,                              currentChildAttrs)                         );                     }                 });             }         });     }     ... }  //A skeletal Channel implementation. public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {     private final DefaultChannelPipeline pipeline;     ...     //Creates a new instance.     protected AbstractChannel(Channel parent) {         this.parent = parent;         id = newId();         unsafe = newUnsafe();         pipeline = newChannelPipeline();     }          protected DefaultChannelPipeline newChannelPipeline() {         return new DefaultChannelPipeline(this);     }      @Override     public ChannelPipeline pipeline() {         return pipeline;     }     ... }  //The default ChannelPipeline implementation.   //It is usually created by a Channel implementation when the Channel is created. public class DefaultChannelPipeline implements ChannelPipeline {     final AbstractChannelHandlerContext head;     final AbstractChannelHandlerContext tail;     ...     protected DefaultChannelPipeline(Channel channel) {         ...         tail = new TailContext(this);         head = new HeadContext(this);         head.next = tail;         tail.prev = head;     }          @Override     public final ChannelPipeline addLast(ChannelHandler... handlers) {         return addLast(null, handlers);     }      @Override     public final ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {         ...         for (ChannelHandler h: handlers) {             if (h == null) break;             addLast(executor, null, h);         }         return this;     }          @Override     public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {         final AbstractChannelHandlerContext newCtx;         synchronized (this) {             checkMultiplicity(handler);             newCtx = newContext(group, filterName(name, handler), handler);             addLast0(newCtx);             ...         }         ...     }          //往Pipeline中添加ChannelHandler处理器     private void addLast0(AbstractChannelHandlerContext newCtx) {         AbstractChannelHandlerContext prev = tail.prev;         newCtx.prev = prev;         newCtx.next = tail;         prev.next = newCtx;         tail.prev = newCtx;     }     ... }

然后，新连接接入调用到服务端Channel的Pipeline的fireChannelRead()方法时，便会触发调用ServerBootstrapAcceptor处理器的channelRead()方法。最终会调用NioEventLoop的register()方法注册这个新连接Channel，即给新连接Channel绑定一个Reactor线程。

//The default ChannelPipeline implementation.   //It is usually created by a Channel implementation when the Channel is created. public class DefaultChannelPipeline implements ChannelPipeline {     final AbstractChannelHandlerContext head;     final AbstractChannelHandlerContext tail;     ...          protected DefaultChannelPipeline(Channel channel) {         ...         tail = new TailContext(this);         head = new HeadContext(this);         head.next = tail;         tail.prev = head;     }          @Override     public final ChannelPipeline fireChannelRead(Object msg) {         //从Pipeline的第一个HeadContext处理器开始调用         AbstractChannelHandlerContext.invokeChannelRead(head, msg);         return this;     }          final class HeadContext extends AbstractChannelHandlerContext implements ChannelOutboundHandler, ChannelInboundHandler {         ...         @Override         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {             //调用AbstractChannelHandlerContext的fireChannelRead()方法             ctx.fireChannelRead(msg);         }               @Override         public ChannelHandler handler() {             return this;         }         ...     }     ... }  abstract class AbstractChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext, ResourceLeakHint {     ...     static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {         final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);         EventExecutor executor = next.executor();         if (executor.inEventLoop()) {             next.invokeChannelRead(m);         } else {             executor.execute(new Runnable() {                 @Override                 public void run() {                     next.invokeChannelRead(m);                 }             });         }     }      private void invokeChannelRead(Object msg) {         if (invokeHandler()) {             try {                 //首先调用的是Pipeline的第一个处理器HeadContext的channelRead()方法                 //注意：HeadContext继承了AbstractChannelHandlerContext                 ((ChannelInboundHandler) handler()).channelRead(this, msg);             } catch (Throwable t) {                 notifyHandlerException(t);             }         } else {             fireChannelRead(msg);         }     }          @Override     public ChannelHandlerContext fireChannelRead(final Object msg) {         invokeChannelRead(findContextInbound(), msg);         return this;     }          private AbstractChannelHandlerContext findContextInbound() {         AbstractChannelHandlerContext ctx = this;         do {             //注意：HeadContext继承了AbstractChannelHandlerContext             //所以如果this是HeadContext，那么这里会获取下一个节点ServerBootstrapAcceptor             ctx = ctx.next;         } while (!ctx.inbound);         return ctx;     }     ... }  public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {     ...     private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter {         private final EventLoopGroup childGroup;         private final ChannelHandler childHandler;         private final Entry<ChannelOption<?>, Object>[] childOptions;         private final Entry<AttributeKey<?>, Object>[] childAttrs;         ...         //channelRead()方法在新连接接入时被调用         @Override         @SuppressWarnings("unchecked")         public void channelRead(ChannelHandlerContext ctx, Object msg) {             final Channel child = (Channel) msg;             //1.给新连接的Channel添加用户自定义的Handler处理器             //这里的childHandler其实是一个特殊的Handler: ChannelInitializer             child.pipeline().addLast(childHandler);             //2.设置ChannelOption，主要和TCP连接一些底层参数及Netty自身对一个连接的参数有关             for (Entry<ChannelOption<?>, Object> e: childOptions) {                 if (!child.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) {                     logger.warn("Unknown channel option: " + e);                 }             }             //3.设置新连接Channel的属性             for (Entry<AttributeKey<?>, Object> e: childAttrs) {                 child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());             }             //4.绑定Reactor线程             //childGroup是一个NioEventLoopGroup，所以下面会调用其父类的register()方法             childGroup.register(child);         }         ...     }     ... }  // MultithreadEventLoopGroup implementations which is used for NIO Selector based Channels. public class NioEventLoopGroup extends MultithreadEventLoopGroup {     ...     ... }  //Abstract base class for EventLoopGroup implementations that handles their tasks with multiple threads at the same time. public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {     ...     @Override     public ChannelFuture register(Channel channel) {         //最终会调用NioEventLoop的register()方法注册这个新连接Channel         return next().register(channel);     }          @Override     public EventLoop next() {         //获取一个NioEventLoop         return (EventLoop) super.next();     }     ... }  //Abstract base class for EventExecutorGroup implementations that handles their tasks with multiple threads at the same time. public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {     private final EventExecutorChooserFactory.EventExecutorChooser chooser;     ...     //Create a new instance.     protected MultithreadEventExecutorGroup(int nThreads, ThreadFactory threadFactory, Object... args) {         this(nThreads, threadFactory == null ? null : new ThreadPerTaskExecutor(threadFactory), args);     }          //Create a new instance.     protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {         this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);     }          protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) {         ...         children = new EventExecutor[nThreads];         for (int i = 0; i < nThreads; i ++) {             children[i] = newChild(executor, args);             ...         }         //创建chooser         chooser = chooserFactory.newChooser(children);         ...     }          @Override     public EventExecutor next() {         //调用chooser的next()方法获得一个NioEventLoop         return chooser.next();     }     ... }  public final class DefaultEventExecutorChooserFactory implements EventExecutorChooserFactory {     public static final DefaultEventExecutorChooserFactory INSTANCE = new DefaultEventExecutorChooserFactory();     private DefaultEventExecutorChooserFactory() { }      @SuppressWarnings("unchecked")     @Override     public EventExecutorChooser newChooser(EventExecutor[] executors) {         if (isPowerOfTwo(executors.length)) {             return new PowerOfTowEventExecutorChooser(executors);         } else {             return new GenericEventExecutorChooser(executors);         }     }      private static boolean isPowerOfTwo(int val) {         return (val & -val) == val;     }      private static final class PowerOfTowEventExecutorChooser implements EventExecutorChooser {         private final AtomicInteger idx = new AtomicInteger();         private final EventExecutor[] executors;                  PowerOfTowEventExecutorChooser(EventExecutor[] executors) {             this.executors = executors;         }                  @Override         public EventExecutor next() {             return executors[idx.getAndIncrement() & executors.length - 1];         }     }      private static final class GenericEventExecutorChooser implements EventExecutorChooser {         private final AtomicInteger idx = new AtomicInteger();         private final EventExecutor[] executors;                  GenericEventExecutorChooser(EventExecutor[] executors) {             this.executors = executors;         }                  @Override         public EventExecutor next() {             return executors[Math.abs(idx.getAndIncrement() % executors.length)];         }     } }

(4)服务端Channel处理新连接的步骤

ServerBootstrapAcceptor处理新连接的步骤：

 

一.给客户端Channel添加childHandler

给客户端Channel添加childHandler也就是将用户自定义的childHandler添加到新连接的pipeline里。

 

pipeline.fireChannelRead(NioSocketChannel)最终会调用到ServerBootstrapAcceptor的channelRead()方法，而且这个channelRead()方法一上来就会把入参的msg强制转换为Channel。

 

拿到新连接的Channel后就可以拿到其对应的Pipeline，这个Pipeline是在调用AbstractChannel构造方法时创建的。于是可以将用户代码中的childHandler添加到Pipeline中，而childHandler其实就是用户代码中的ChannelInitializer。所以新连接Channel的Pipeline的构成是：Head -> ChannelInitializer -> Tail。

 

二.设置客户端Channel的options和attr

所设置的childOptions和childAttrs也是在用户代码中设置的，这些设置项最终会传递到ServerBootstrapAcceptor的channelRead()方法中进行具体设置。

 

三.选择NioEventLoop绑定客户端Channel

childGroup.register(child)中的childGroup就是用户代码里创建的workerNioEventLoopGroup。NioEventLoopGroup的register()方法会调用next()由其父类通过线程选择器chooser返回一个NioEventLoop。所以childGroup.register(child)最终会调用到NioEventLoop的register()方法，这和注册服务端Channel时调用config().group().register(channel)一样。

 

(5)总结

服务端Channel在检测到新连接并且创建完客户端Channel后，会通过服务端Channel的Pipeline的一个处理器ServerBootstrapAcceptor做一些处理。这些处理包括：给客户端Channel的Pipeline添加childHandler处理器、设置客户端Channel的options和attrs、调用线程选择器chooser选择一个NioEventLoop进行绑定。绑定时会将该客户端Channel注册到NioEventLoop的Selector上，此时还不会关心事件。

 

7.新连接接入之注册Selector和注册读事件

NioEventLoop的register()方法是由其父类SingleThreadEventLoop实现的，并最终调用到AbstractChannel的内部类AbstractUnsafe的register0()方法。

 

步骤一：注册Selector

和服务端启动过程一样，先调用AbstractNioChannel的doRegister()方法进行注册。其中javaChannel().register()会将新连接NioSocketChannel绑定到Reactor线程的Selector上，这样后续这个新连接NioSocketChannel所有的事件都由绑定的Reactor线程的Selector来轮询。

 

步骤二：配置自定义Handler

此时新连接NioSocketChannel的Pipeline中有三个Handler：Head -> ChannelInitializer -> Tail。invokeHandlerAddedIfNeeded()最终会调用ChannelInitializer的handlerAdded()方法。

 

步骤三：传播ChannelRegistered事件

pipeline.fireChannelRegistered()会把新连接的注册事件从HeadContext开始往下传播，调用每一个ChannelHandler的channelRegistered()方法。

 

步骤四：注册读事件

接着还会传播ChannelActive事件。传播完ChannelActive事件后，便会继续调用HeadContetx的readIfIsAutoRead()方法注册读事件。由于创建NioSocketChannel时已将SelectionKey.OP_READ的事件代码保存到其成员变量中，所以AbstractNioChannel的doBeginRead()方法，就可以将SelectionKey.OP_READ事件注册到Selector中完成读事件的注册。

//A skeletal Channel implementation. public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {     private final ChannelId id;     private final Unsafe unsafe;     private final DefaultChannelPipeline pipeline;     private volatile EventLoop eventLoop;     ...          //Creates a new instance.     protected AbstractChannel(Channel parent) {         this.parent = parent;         id = newId();         unsafe = newUnsafe();         pipeline = newChannelPipeline();     }          //Returns a new DefaultChannelPipeline instance.     protected DefaultChannelPipeline newChannelPipeline() {         return new DefaultChannelPipeline(this);     }      //Unsafe implementation which sub-classes must extend and use.     protected abstract class AbstractUnsafe implements Unsafe {         ...         @Override         public final void register(EventLoop eventLoop, final ChannelPromise promise) {             ...             //绑定事件循环器，即绑定一个NioEventLoop到该Channel上             AbstractChannel.this.eventLoop = eventLoop;             //注册Selector，并启动一个NioEventLoop             if (eventLoop.inEventLoop()) {                 register0(promise);             } else {                 ...                 //通过启动这个NioEventLoop线程来调用register0()方法将这个Channel注册到Selector上                 eventLoop.execute(new Runnable() {                     @Override                     public void run() {                         register0(promise);                     }                 });                 ...             }         }          private void register0(ChannelPromise promise) {             ...             boolean firstRegistration = this.neverRegistered;             //1.调用JDK底层注册Channel到Selector上             doRegister();             this.neverRegistered = false;             this.registered = true;             //2.配置自定义Handler             this.pipeline.invokeHandlerAddedIfNeeded();             safeSetSuccess(promise);             //3.传播channelRegisterd事件             this.pipeline.fireChannelRegistered();             //4.注册读事件             if (isActive()) {                 if (firstRegistration) {                     //会进入这个方法，传播完ChannelActive事件后，再注册读事件                     this.pipeline.fireChannelActive();                 } else if (config().isAutoRead()) {                     beginRead();                 }             }             ...         }               @Override         public final void beginRead() {             ...             //调用AbstractNioChannel实现的doBeginRead()方法             doBeginRead();             ...         }         ...     }          //Is called after the Channel is registered with its EventLoop as part of the register process.     //Sub-classes may override this method     protected void doRegister() throws Exception {         // NOOP     }          //Schedule a read operation.     protected abstract void doBeginRead() throws Exception;          @Override     public Channel read() {         //调用DefaultChannelPipeline的read()方法         pipeline.read();         return this;     }     ... }  //Abstract base class for Channel implementations which use a Selector based approach. public abstract class AbstractNioChannel extends AbstractChannel {     private final SelectableChannel ch;//这是NIO中的Channel     protected final int readInterestOp;     volatile SelectionKey selectionKey;     ...          //Create a new instance     //@param parent，the parent Channel by which this instance was created. May be null.     //@param ch，he underlying SelectableChannel on which it operates     //@param readInterestOp，the ops to set to receive data from the SelectableChannel     protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {         super(parent);         //NioServerSocketChannel.newSocket()方法通过JDK底层创建的Channel对象会被缓存在其父类AbstractNioChannel的变量ch中         //可以通过NioServerSocketChannel.javaChannel()方法获取其父类AbstractNioChannel的变量ch         this.ch = ch;         this.readInterestOp = readInterestOp;         ...         //设置Channel对象为非阻塞模式         ch.configureBlocking(false);         ...     }          @Override     protected void doRegister() throws Exception {         boolean selected = false;         for (;;) {             ...             //首先获取前面创建的JDK底层NIO的Channel，然后调用JDK底层NIO的register()方法，             //将this也就是NioServerSocketChannel对象当作attachment绑定到JDK的Selector上；             //这样绑定是为了后续从Selector拿到对应的事件后，可以把Netty领域的Channel拿出来；             //而且注册的ops值是0，表示此时还不关注任何事件；             selectionKey = javaChannel().register(eventLoop().selector, 0, this);             return;             ...         }     }              protected SelectableChannel javaChannel() {         return ch;     }          @Override     protected void doBeginRead() throws Exception {         // Channel.read() or ChannelHandlerContext.read() was called         final SelectionKey selectionKey = this.selectionKey;         if (!selectionKey.isValid()) {             return;         }          readPending = true;         final int interestOps = selectionKey.interestOps();         if ((interestOps & readInterestOp) == 0) {             //将SelectionKey.OP_READ读事件注册到Selector上，表示这个客户端Channel可以处理读事件了             selectionKey.interestOps(interestOps | readInterestOp);         }     }     ... }  //The default ChannelPipeline implementation.   //It is usually created by a Channel implementation when the Channel is created. public class DefaultChannelPipeline implements ChannelPipeline {     final AbstractChannelHandlerContext head;     final AbstractChannelHandlerContext tail;     ...     @Override     public final ChannelPipeline fireChannelActive() {         //调用HeadContext的channelActive()方法         AbstractChannelHandlerContext.invokeChannelActive(head);         return this;     }          @Override     public final ChannelPipeline read() {         //从TailContext开始，最终会调用到HeadContext的read()方法         tail.read();         return this;     }          final class HeadContext extends AbstractChannelHandlerContext implements ChannelOutboundHandler, ChannelInboundHandler {         private final Unsafe unsafe;         ...         @Override         public void channelActive(ChannelHandlerContext ctx) throws Exception {             ctx.fireChannelActive();//传播ChannelActive事件             readIfIsAutoRead();         }                 private void readIfIsAutoRead() {             if (channel.config().isAutoRead()) {                 //调用AbstractChannel的read()方法                 channel.read();             }         }                 @Override         public void read(ChannelHandlerContext ctx) {             //调用AbstractChannel.AbstractUnsafe的beginRead()方法             unsafe.beginRead();         }         ...     } }

 

8.注册Reactor线程总结

一.首先当boss Reactor线程在检测到有ACCEPT事件之后，会创建JDK底层的Channel。

 

二.然后使用一个NioSocketChannel包装JDK底层的Channel，把用户设置的ChannelOption、ChannelAttr、ChannelHandler都设置到该NioSocketChannel中。

 

三.接着从worker Reactor线程组中，也就是worker NioEventLoopGroup中，通过线程选择器chooser选择一个NioEventLoop出来。

 

四.最后把NioSocketChannel包装的JDK底层Channel当作key，自身NioSocketChannel当作attachment，注册到NioEventLoop对应的Selector上。这样后续有读写事件发生时，就可以从底层Channel直接获得attachment即NioSocketChannel来进行读写数据的逻辑处理。

 

9.新连接接入总结

新连接接入整体可以分为两部分：一是检测新连接，二是注册Reactor线程。

 

一.首先在Netty服务端的Channel(也就是NioServerSocketChannel)绑定的NioEventLoop(也就是boss线程)中，轮询到ACCEPT事件。

 

二.然后调用JDK的服务端Channel的accept()方法获取一个JDK的客户端Channel，并且将其封装成Netty的客户端Channel(即NioSocketChannel)。

 

三.封装过程中会创建这个NioSocketChannel一系列的组件，如unsafe组件和pipeline组件。unsafe组件主要用于进行Channel的读写，pipeline组件主要用于处理Channel数据的业务逻辑。

 

四.接着Netty服务端Channel的Pipeline的一个处理器ServerBootstrapAcceptor，会给当前Netty客户端Channel分配一个NioEventLoop并将客户端Channel绑定到Selector上。

 

五.最后会传播ChannelRegistered事件和ChannelActive事件，并将客户端Channel的读事件注册到Selector上。

 

至此，新连接NioSocketChannel便可以开始正常读写数据了。