一、介绍

Channelpipeline是 Channelhandler的容器,它负责 Channelhandler的管理和事件拦截与调度。

事件如何在流水线中流动如图下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40



*                                                 I/O Request
*                                            via {@link Channel} or
*                                        {@link ChannelHandlerContext}
*                                                      |
*  +---------------------------------------------------+---------------+
*  |                           ChannelPipeline         |               |
*  |                                                  \|/              |
*  |    +---------------------+            +-----------+----------+    |
*  |    | Inbound Handler  N  |            | Outbound Handler  1  |    |
*  |    +----------+----------+            +-----------+----------+    |
*  |              /|\                                  |               |
*  |               |                                  \|/              |
*  |    +----------+----------+            +-----------+----------+    |
*  |    | Inbound Handler N-1 |            | Outbound Handler  2  |    |
*  |    +----------+----------+            +-----------+----------+    |
*  |              /|\                                  .               |
*  |               .                                   .               |
*  | ChannelHandlerContext.fireIN_EVT() ChannelHandlerContext.OUT_EVT()|
*  |        [ method call]                       [method call]         |
*  |               .                                   .               |
*  |               .                                  \|/              |
*  |    +----------+----------+            +-----------+----------+    |
*  |    | Inbound Handler  2  |            | Outbound Handler M-1 |    |
*  |    +----------+----------+            +-----------+----------+    |
*  |              /|\                                  |               |
*  |               |                                  \|/              |
*  |    +----------+----------+            +-----------+----------+    |
*  |    | Inbound Handler  1  |            | Outbound Handler  M  |    |
*  |    +----------+----------+            +-----------+----------+    |
*  |              /|\                                  |               |
*  +---------------+-----------------------------------+---------------+
*                  |                                  \|/
*  +---------------+-----------------------------------+---------------+
*  |               |                                   |               |
*  |       [ Socket.read() ]                    [ Socket.write() ]     |
*  |                                                                   |
*  |  Netty Internal I/O Threads (Transport Implementation)            |
*  +-------------------------------------------------------------------+
* 

消息的读取和发送处理全流程描述如下:

1)底层的Socket.read()方法读取ByteBuf,触发Channelread事件,由IO线程 Nioeventloop调用 Channelpipeline的fireChannelread(Object msg)方法,将消息 (ByteBuf)传输到 Channelpipeline中,将消息依次被 Inbound Handler 1、 Inbound Handler 2、 Inbound Handler N拦截和处理,在这个过程中, 任何Channelhandler都可以中断当前的流程,结束消息的传递。
2)调用ChannelHandlerContext的wrte方法发送消息,将消息依次被Outbound Handler 1,Outbound Handler 2 ,Outbound Handler M 拦截和处理,最终交给Socket.write()发送消息；

将事件转发给下一个处理程序

正如上图所看到的那样，处理程序必须调用ChannelHandlerContext中的事件传播方法才能将事件转发到其下一个处理程序。这些方法包括：

Inbound 事件传播方法:
ChannelHandlerContext.fireChannelRegistered() Channel注册事件
ChannelHandlerContext.fireChannelActive() TCP链路建立成功, Channel激活事件;
ChannelHandlerContext.fireChannelRead(Object) 读事件
ChannelHandlerContext.fireChannelReadComplete() 读操作完成通知事件;
ChannelHandlerContext.fireExceptionCaught(Throwable) 异常通知事件;
ChannelHandlerContext.fireUserEventTriggered(Object) 用户自定义事件
ChannelHandlerContext.fireChannelWritabilityChanged() Channel的可写状态变化通知事件;
ChannelHandlerContext.fireChannelInactive() TCP连接关闭,链路不可用通知事件。
ChannelHandlerContext.fireChannelUnregistered() Channel从EventLoop中注销
Outbound 事件传播方法:
ChannelOutboundInvoker.bind(SocketAddress, ChannelPromise)
ChannelOutboundInvoker.connect(SocketAddress, SocketAddress, ChannelPromise)
ChannelOutboundInvoker.write(Object, ChannelPromise)
ChannelHandlerContext.flush()
ChannelHandlerContext.read()
ChannelOutboundInvoker.disconnect(ChannelPromise)
ChannelOutboundInvoker.close(ChannelPromise)
ChannelOutboundInvoker.deregister(ChannelPromise)

线程安全：

Channelpipeline是线程安全的,这意味着N个业务线程可以并发地操作Channelpipeline而不存在多线程并发问题。但是, Channelhandler却不是线程安全的,这意味着尽管 Channelpipeline是线程安全的,但是用户仍然需要自己保证Channelhandler的线程安全，ChannelHandler可以随时添加或删除；

uml关系如图下：

二、 DefaultChannelPipeline源码分析

1.数据结构

以双链表保存ChannelHandler的数据
head和tail变量，控制往下处理或者往上处理

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34


public class DefaultChannelPipeline implements ChannelPipeline {


    private static final String HEAD_NAME = generateName0(HeadContext.class);
    private static final String TAIL_NAME = generateName0(TailContext.class);

    private static final FastThreadLocal<Map<Class<?>, String>> nameCaches =  
            new FastThreadLocal<Map<Class<?>, String>>() {
        @Override
        protected Map<Class<?>, String> initialValue() throws Exception {
            return new WeakHashMap<Class<?>, String>();
        }
    };

    private static final AtomicReferenceFieldUpdater<DefaultChannelPipeline, MessageSizeEstimator.Handle> ESTIMATOR =
            AtomicReferenceFieldUpdater.newUpdater(
                    DefaultChannelPipeline.class, MessageSizeEstimator.Handle.class, "estimatorHandle");
    //AbstractChannelHandlerContext中有prev,next属性形成双链表；Inbound和Outbound事件都在在这个抽象类实现；
    final AbstractChannelHandlerContext head;  //头节点，
    final AbstractChannelHandlerContext tail;  //尾节点 

    private final Channel channel;
    private final ChannelFuture succeededFuture;
    private final VoidChannelPromise voidPromise;
    private final boolean touch = ResourceLeakDetector.isEnabled();
    //线程池中的线程映射，记住这个映射是为了保证执行任务时使用同一个线程
    private Map<EventExecutorGroup, EventExecutor> childExecutors;
    private volatile MessageSizeEstimator.Handle estimatorHandle; //负责估计消息的大小
    private final boolean firstRegistration = true;//对应Channel首次注册到EventLoop 
    //未注册，添加或者删除ChannelHandler预备存放地方，单链表存储
    private PendingHandlerCallback pendingHandlerCallbackHead;
    //Channel被注册，就设置为true。一旦设置为true，值将不会改变。
    private boolean registered;
}    

2.构造

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


    protected DefaultChannelPipeline(Channel channel) {
        this.channel = ObjectUtil.checkNotNull(channel, "channel");
        succeededFuture = new SucceededChannelFuture(channel, null);
        voidPromise =  new VoidChannelPromise(channel, true);

        tail = new TailContext(this);//头节点 
        head = new HeadContext(this);//尾节点 
        //头节点和尾节点相互链接
        head.next = tail;
        tail.prev = head;
    }

3.addFirst 添加ChannelHandler

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36


    @Override
    public final ChannelPipeline addFirst(EventExecutorGroup group, String name, ChannelHandler handler) {
        final AbstractChannelHandlerContext newCtx;
        synchronized (this) {
            // 1.验证未被@Sharable注解时,一个实例是否重复使用
            checkMultiplicity(handler);
            // 2.名称重复样子，以及name为空，自动生成名字，生成名字规则按类名
            name = filterName(name, handler);
            // 3.创建AbstractChannelHandlerContext
            newCtx = newContext(group, name, handler);
            // 4.添加双链表中
            addFirst0(newCtx);

            // 5.未注册，PendingHandlerCallback任务添加到pendingHandlerCallbackHead链表中；
            //   分添加(PendingHandlerCallback)或者删除任务(PendingHandlerRemovedTask)
            if (!registered) {
                newCtx.setAddPending();  //设置AbstractChannelHandlerContext中状态，INIT改成ADD_PENDING
                callHandlerCallbackLater(newCtx, true);
                return this;
            }
            // 6.当前线程不是EventLoop线程，是直接callHandlerAdded0；不是添加任务执行callHandlerAdded0
            EventExecutor executor = newCtx.executor();
            if (!executor.inEventLoop()) { //当前线程不是EventLoop线程
                newCtx.setAddPending();
                executor.execute(new Runnable() {
                    @Override
                    public void run() {
                        callHandlerAdded0(newCtx);
                    }
                });
                return this;
            }
        }
        callHandlerAdded0(newCtx);
        return this;
    }

(1).checkMultiplicity 验证@Sharable注解，未被@Sharable注解时，指定ChannelHandler对象ChannelPipeline只能添加一次；被@Sharable注解时，可以添加多次；

handler继承ChannelHandlerAdapter,才能做下面判断
- 类注解@Sharable，一个实例添加多次添加
- 没有类注解@Sharable,一个实例只能添加一次，以added做验证

1
2
3
4
5
6
7
8
9


 private static void checkMultiplicity(ChannelHandler handler) {
        if (handler instanceof ChannelHandlerAdapter) {
            ChannelHandlerAdapter h = (ChannelHandlerAdapter) handler;
            if (!h.isSharable() && h.added) { //验证ChannelHandler对象未被@Sharable注解时，ChannelPipeline只能添加一次；
                throw new ChannelPipelineException(...);
            }
            h.added = true;
        }
    }

(2).filterName name为空自动生成名称，name不为空，验证名称是否重复

参数name为NULL,自动生成名字
- 按类名称自动生成名称，有相同名字后面添加#1
检查重复名称

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35


   private String filterName(String name, ChannelHandler handler) {
        if (name == null) {  //名称为空，自动生成名称，
            return generateName(handler);
        }
        checkDuplicateName(name); //检查重复名称
        return name;
    }
    
    //按handler类自动生成名字
     private String generateName(ChannelHandler handler) {
          Map<Class<?>, String> cache = nameCaches.get();
          Class<?> handlerType = handler.getClass();
          String name = cache.get(handlerType);/先从缓存中获取名称
          if (name == null) {
              name = generateName0(handlerType); //按类名称自动生成名称
              cache.put(handlerType, name);      //保存在缓存中
          }
          //用户不太可能放置多个相同类型的处理程序，但要确保避免任何名称冲突。请注意，我们不会缓存这里生成的名称。
          if (context0(name) != null) {
              String baseName = name.substring(0, name.length() - 1); // Strip the trailing '0'.
              //有相同名称，baseName把后面#0的0累加生成新名称；
              for (int i = 1;; i ++) {
                  String newName = baseName + i;
                  if (context0(newName) == null) {
                      name = newName;   
                      break;
                  }
              }
          }
          return name;
      }  
      //生成名称规则 handlerType=com.zp.HelloChannelHandler 形成名称=HelloChannelHandler#0
      private static String generateName0(Class<?> handlerType) {
          return StringUtil.simpleClassName(handlerType) + "#0";
      } 

(3).newContext 创建AbstractChannelHandlerContext，这个类在后面具体分析；

ChannelHandler包装成ChannelHandlerContext;

1
2
3


    private AbstractChannelHandlerContext newContext(EventExecutorGroup group, String name, ChannelHandler handler) {
        return new DefaultChannelHandlerContext(this, childExecutor(group), name, handler);
    }

a.childExecutor(group)

根据ChannelOption.SINGLE_EVENTEXECUTOR_PER_GROUP配置，按线程组是否公用相同的线程

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20


   private EventExecutor childExecutor(EventExecutorGroup group) {
       if (group == null) {
           return null;
       }
       //单独执行任务
       Boolean pinEventExecutor = channel.config().getOption(ChannelOption.SINGLE_EVENTEXECUTOR_PER_GROUP);
       if (pinEventExecutor != null && !pinEventExecutor) {
           return group.next();
       }
       Map<EventExecutorGroup, EventExecutor> childExecutors = this.childExecutors;
       if (childExecutors == null) {
           childExecutors = this.childExecutors = new IdentityHashMap<EventExecutorGroup, EventExecutor>(4);
       }
       EventExecutor childExecutor = childExecutors.get(group);
       if (childExecutor == null) {
           childExecutor = group.next();
           childExecutors.put(group, childExecutor);
       }
       return childExecutor;
   } 

(4).addFirst0 拼接到head后面

链表添加

1
2
3
4
5
6
7


  private void addFirst0(AbstractChannelHandlerContext newCtx) {
         AbstractChannelHandlerContext nextCtx = head.next;
         newCtx.prev = head;
         newCtx.next = nextCtx;
         head.next = newCtx;
         nextCtx.prev = newCtx;
     }

(5).callHandlerCallbackLater registered为false，把回调任务添加到pendingHandlerCallbackHead链表中；

pendingHandlerCallbackHead添加任务，单链表结构
分两种类型PendingHandlerCallbackPendingHandlerCallback和PendingHandlerRemovedTask
- 调用callHandlerAdded0方法以及callHandlerAdded0
Channel注册调用channelRegistered方法，处理回调任务链表集合；处理方法invokeHandlerAddedIfNeeded

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15


    private void callHandlerCallbackLater(AbstractChannelHandlerContext ctx, boolean added) {
        assert !registered;

        PendingHandlerCallback task = added ? new PendingHandlerCallbackPendingHandlerCallback(ctx) : new PendingHandlerRemovedTask(ctx);
        PendingHandlerCallback pending = pendingHandlerCallbackHead;
        if (pending == null) {
            pendingHandlerCallbackHead = task;
        } else {
            // Find the tail of the linked-list.
            while (pending.next != null) {
                pending = pending.next;
            }
            pending.next = task;
        }
    }

a.PendingHandlerAddedTask 和PendingHandlerRemovedTask源码分析

PendingHandlerCallback抽象类是PendingHandlerAddedTask 和PendingHandlerRemovedTask父类；

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


   private abstract static class PendingHandlerCallback implements Runnable {
        final AbstractChannelHandlerContext ctx;
        PendingHandlerCallback next; //单链表实现；保存集合

        PendingHandlerCallback(AbstractChannelHandlerContext ctx) {
            this.ctx = ctx;
        }

        abstract void execute();
    }

PendingHandlerAddedTask 等待添加任务

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28


 private final class PendingHandlerAddedTask extends PendingHandlerCallback {

        PendingHandlerAddedTask(AbstractChannelHandlerContext ctx) {
            super(ctx);
        }

        @Override
        public void run() {
            callHandlerAdded0(ctx);
        }

        @Override
        void execute() {
            EventExecutor executor = ctx.executor();
            if (executor.inEventLoop()) {
                callHandlerAdded0(ctx);
            } else {
                try {
                    executor.execute(this);
                } catch (RejectedExecutionException e) {
                    ...
                    remove0(ctx);
                    ctx.setRemoved();
                }
            }
        }
    }

PendingHandlerRemovedTask 等待删除任务

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26


    private final class PendingHandlerRemovedTask extends PendingHandlerCallback {

        PendingHandlerRemovedTask(AbstractChannelHandlerContext ctx) {
            super(ctx);
        }

        @Override
        public void run() {
            callHandlerRemoved0(ctx);
        }

        @Override
        void execute() {
            EventExecutor executor = ctx.executor();
            if (executor.inEventLoop()) {
                callHandlerRemoved0(ctx);
            } else {
                try {
                    executor.execute(this);
                } catch (RejectedExecutionException e) {
                   ...
                    ctx.setRemoved();
                }
            }
        }
    }

b.invokeHandlerAddedIfNeeded 处理理回调任务链表pendingHandlerCallbackHead集合

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26


 final void invokeHandlerAddedIfNeeded() {
        assert channel.eventLoop().inEventLoop();
        if (firstRegistration) {
            firstRegistration = false;
            //我们现在注册到EventLoop。现在是时候调用ChannelHandlers的回调了，这是在注册完成之前添加的。
            callHandlerAddedForAllHandlers();
        }
    }
    private void callHandlerAddedForAllHandlers() {
        final PendingHandlerCallback pendingHandlerCallbackHead;
        synchronized (this) {
            assert !registered;

            // This Channel itself was registered.
            registered = true;

            pendingHandlerCallbackHead = this.pendingHandlerCallbackHead;
            // Null out so it can be GC'ed.
            this.pendingHandlerCallbackHead = null;
        }
        PendingHandlerCallback task = pendingHandlerCallbackHead;
        while (task != null) {
            task.execute(); //执行任务
            task = task.next;
        }
    }    

(6).callHandlerAdded0 通知添加完成处理两个方法ChannelHandler.handlerAdded和AbstractChannelHandlerContext.setAddComplete

通知ChannelHandler.handlerAdded方法
通知AbstractChannelHandlerContext.setAddComplete方法，ADD_COMPLETE

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25


    private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
        try {
            ctx.handler().handlerAdded(ctx); //通知ChannelHandler.handlerAdded方法
            ctx.setAddComplete();            //通知AbstractChannelHandlerContext.setAddComplete方法，ADD_COMPLETE
        } catch (Throwable t) { //通知异常处理
            boolean removed = false;
            try {
                remove0(ctx);
                try {
                    ctx.handler().handlerRemoved(ctx);//通知ChannelHandler.handlerRemoved方法
                } finally {
                    ctx.setRemoved();                  //通知ChannelHandler.setRemoved方法 REMOVE_COMPLETE
                }
                removed = true;
            } catch (Throwable t2) {
                ...
            }

            if (removed) {
                fireExceptionCaught(new ChannelPipelineException(...));//异常通知事件
            } else {
                fireExceptionCaught(new ChannelPipelineException(...));//异常通知事件
            }
        }
    }

a.remove0 移除AbstractChannelHandlerContext链表中

1
2
3
4
5
6


    private static void remove0(AbstractChannelHandlerContext ctx) {
        AbstractChannelHandlerContext prev = ctx.prev;
        AbstractChannelHandlerContext next = ctx.next;
        prev.next = next;
        next.prev = prev;
    }

(7).callHandlerRemoved0 删除完成通知

remove调用方法

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12


private void callHandlerRemoved0(final AbstractChannelHandlerContext ctx) {
       // Notify the complete removal.
       try {
           try {
               ctx.handler().handlerRemoved(ctx);//通知ChannelHandler.handlerRemoved方法
           } finally {
               ctx.setRemoved();                 //通知ChannelHandler.setRemoved方法
           }
       } catch (Throwable t) {
           fireExceptionCaught(new ChannelPipelineException(...));
       }
   }

4.addLast,addBefore都类同，双链表操作

5.remove 删除指定AbstractChannelHandlerContext

6.replace 替换指定AbstractChannelHandlerContext

6.destroy pipeline中的所有节点销毁，顺序由尾部向头部并触发handlerRemoved（）

7.Inbound 事件方法

head触发，判断事件执行器是否当前线程，是立即执行head的事件，不是，提交任务执行；

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


  @Override
    public final ChannelPipeline fireChannelRegistered() {
        AbstractChannelHandlerContext.invokeChannelRegistered(head);
        return this;
    }

    @Override
    public final ChannelPipeline fireChannelUnregistered() {
        AbstractChannelHandlerContext.invokeChannelUnregistered(head);
        return this;
    }

8.Outbound 事件方法

tail执行,tail是Inbound事件；

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


  @Override
    public final ChannelFuture bind(SocketAddress localAddress) {
        return tail.bind(localAddress);
    }

    @Override
    public final ChannelFuture connect(SocketAddress remoteAddress) {
        return tail.connect(remoteAddress);
    }

 

tail就是TailContext对象，TailContext实现AbstractChannelHandlerContext抽象类；具有AbstractChannelHandlerContext特性；

三、 AbstractChannelHandlerContext源码分析

AbstractChannelHandlerContext抽象类继承ChannelHandlerContext，具有ChannelHandlerContext.Inbound和ChannelHandlerContext.Outbound事件特性；

AbstractChannelHandlerContext的UML图

1.数据结构

双链表结构
handlerState 状态控制是否执行

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38


abstract class AbstractChannelHandlerContext extends DefaultAttributeMap
        implements ChannelHandlerContext, ResourceLeakHint {

    //双向链表
    volatile AbstractChannelHandlerContext next;//下节点
    volatile AbstractChannelHandlerContext prev;//前节点
    //CAS 修改handlerState属性
    private static final AtomicIntegerFieldUpdater<AbstractChannelHandlerContext> HANDLER_STATE_UPDATER =
            AtomicIntegerFieldUpdater.newUpdater(AbstractChannelHandlerContext.class, "handlerState");

    //handlerState 几个状态标示

    //等待添加状态
    private static final int ADD_PENDING = 1;

    //添加状态
    private static final int ADD_COMPLETE = 2;

    //删除状态
    private static final int REMOVE_COMPLETE = 3;

    //初始化状态
    private static final int INIT = 0;

    private final boolean inbound;//inbound事件状态
    private final boolean outbound;//outbound事件状态
    private final DefaultChannelPipeline pipeline;
    private final String name;  //ChannelHandler名称
    //如果由EventLoop或给定的EventExecutor的是OrderedEventExecutor的实例，则它是有序的。
    private final boolean ordered;
    
    //如果不使用子执行程序，将被设置为null，否则将被设置为子执行程序。
    final EventExecutor executor;
    private ChannelFuture succeededFuture;
    
    //状态，初始化 INIT
    private final int handlerState = INIT;
}    

2.构造

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


    AbstractChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutor executor, String name,
                                  boolean inbound, boolean outbound) {
        this.name = ObjectUtil.checkNotNull(name, "name");
        this.pipeline = pipeline;
        this.executor = executor;
        this.inbound = inbound;
        this.outbound = outbound;
        //如果由EventLoop或给定的EventExecutor的是OrderedEventExecutor的实例，则它是有序的。
        ordered = executor == null || executor instanceof OrderedEventExecutor;
    }

3.executor 获取事件执行器(就是个线程)

没有指定事件执行器就用Channel的事件执行去

1
2
3
4
5
6
7
8


   @Override
   public EventExecutor executor() {
       if (executor == null) {
           return channel().eventLoop();
       } else {
           return executor;
       }
   }

4.findContextInbound 查找Inbound事件的ChannelHandlerContext

往下个节点查询，inbound为true的ChannelHandlerContext；

1
2
3
4
5
6
7


   private AbstractChannelHandlerContext findContextInbound() {
       AbstractChannelHandlerContext ctx = this;
       do {
           ctx = ctx.next;
       } while (!ctx.inbound);
       return ctx;
   } 

5.Inbound事件

(1).fireChannelRegistered() 触发next的注册事件

分三个步骤执行：

查找next的ChannelHandlerContext
调用next个ChannelHandlerContext#invokeChannelRegistered方法
执行ChannelInboundHandler#channelRegistered

1
2
3
4
5


 @Override
 public ChannelHandlerContext fireChannelRegistered() {
     invokeChannelRegistered(findContextInbound());
     return this;
 } 

a.findContextInbound 查找next的ChannelHandlerContext

往下个节点查询，inbound为true的ChannelHandlerContext；

1
2
3
4
5
6
7


   private AbstractChannelHandlerContext findContextInbound() {
       AbstractChannelHandlerContext ctx = this;
       do {
           ctx = ctx.next;
       } while (!ctx.inbound);
       return ctx;
   } 

b.invokeChannelRegistered 调用next个ChannelHandlerContext#invokeChannelRegistered方法

获取next的事件执行器，是相同线程就立即执行invokeChannelRegistered，不是提交任务执行;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


    static void invokeChannelRegistered(final AbstractChannelHandlerContext next) {
        //获取next的事件执行器，是相同线程就立即执行invokeChannelRegistered，不是提交任务执行;
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeChannelRegistered();
        } else {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    next.invokeChannelRegistered();
                }
            });
        }
    } 

c.invokeChannelRegistered执行ChannelInboundHandler#channelRegistered

验证当前状态，完成或者等待添加执，执行ChannelInboundHandler#channelRegistered
- 发送异常，判断异常栈中没有方法名称等于exceptionCaught，就要调用ChannelInboundHandler.exceptionCaught
  - 等于exceptionCaught，就是异常处理发送异常，不做处理
否，则往next查找执行

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


    private void invokeChannelRegistered() {
        // 验证当前状态，完成或者等待添加执，执行ChannelInboundHandler#channelRegistered
        // handlerState == ADD_COMPLETE || (!ordered && handlerState == ADD_PENDING);
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelRegistered(this);//调用ChannelInboundHandler.channelRegistered
            } catch (Throwable t) {
                //判断异常栈中没有发现方法名称等于exceptionCaught，就要调用ChannelInboundHandler.exceptionCaught
                notifyHandlerException(t);
            }
        } else {
            fireChannelRegistered();//往下查找
        }
    }