Netty学习笔记（番外篇） - ChannelHandler、ChannelPipeline和ChannelHandlerContext的联系 (3)

以上代码实现逻辑是这样的：当创建一个 ChannelHandlerContext 时，会与一个 ChannelHandler 绑定，同时会将传递进来的 ChannelHandler 进行解析，解析当前 ChannelHandler 支持哪些回调方法，并通过位运算得到一个结果保存在 ChannelHandlerContext 的 executionMask 属性里。注意 m.isAnnotationPresent(Skip.class)这里，ChannelHandler 的基类 ChannelInboundHandlerAdapter 和 ChannelOutboundHandlerAdapter 里的回调方法上都有@Skip 注解，当继承了这两个类并重写了某个回调方法后，这个方法上的注解就会被覆盖掉，解析时就会被认为当前 ChannelHandler 支持这个回调方法。
下面是每个回调方法对应的掩码

// ChannelHandlerMask.java final class ChannelHandlerMask { // Using to mask which methods must be called for a ChannelHandler. static final int MASK_EXCEPTION_CAUGHT = 1; static final int MASK_CHANNEL_REGISTERED = 1 << 1; static final int MASK_CHANNEL_UNREGISTERED = 1 << 2; static final int MASK_CHANNEL_ACTIVE = 1 << 3; static final int MASK_CHANNEL_INACTIVE = 1 << 4; static final int MASK_CHANNEL_READ = 1 << 5; static final int MASK_CHANNEL_READ_COMPLETE = 1 << 6; static final int MASK_USER_EVENT_TRIGGERED = 1 << 7; static final int MASK_CHANNEL_WRITABILITY_CHANGED = 1 << 8; static final int MASK_BIND = 1 << 9; static final int MASK_CONNECT = 1 << 10; static final int MASK_DISCONNECT = 1 << 11; static final int MASK_CLOSE = 1 << 12; static final int MASK_DEREGISTER = 1 << 13; static final int MASK_READ = 1 << 14; static final int MASK_WRITE = 1 << 15; static final int MASK_FLUSH = 1 << 16; static final int MASK_ONLY_INBOUND = MASK_CHANNEL_REGISTERED | MASK_CHANNEL_UNREGISTERED | MASK_CHANNEL_ACTIVE | MASK_CHANNEL_INACTIVE | MASK_CHANNEL_READ | MASK_CHANNEL_READ_COMPLETE | MASK_USER_EVENT_TRIGGERED | MASK_CHANNEL_WRITABILITY_CHANGED; private static final int MASK_ALL_INBOUND = MASK_EXCEPTION_CAUGHT | MASK_ONLY_INBOUND; static final int MASK_ONLY_OUTBOUND = MASK_BIND | MASK_CONNECT | MASK_DISCONNECT | MASK_CLOSE | MASK_DEREGISTER | MASK_READ | MASK_WRITE | MASK_FLUSH; private static final int MASK_ALL_OUTBOUND = MASK_EXCEPTION_CAUGHT | MASK_ONLY_OUTBOUND; } 二、ChannelHandler 处理消息

我们以消息读取和写入为例，来看看在 ChannelPipeline 里的各个 ChannelHandler 是如何按照顺序处理消息和事件的。

读取消息

当 Channel 读取到消息后，会在以下地方调用 ChannelPipeline 的 fireChannelRead 方法：

// AbstractNioMessageClient.java private final class NioMessageUnsafe extends AbstractNioUnsafe { // 省略代码 @Override public void read() { // ...... for (int i = 0; i < size; i ++) { readPending = false; pipeline.fireChannelRead(readBuf.get(i)); } // ...... } // 省略代码 } // DefaultChannelPipeline.java // 省略代码 @Override public final ChannelPipeline fireChannelRead(Object msg) { AbstractChannelHandlerContext.invokeChannelRead(head, msg); return this; } // 省略代码

可以看到，通过 AbstractChannelHandlerContext 的 invokeChannelRead 方法，传递 head，从头部开始触发读取事件。

// AbstractChannelHandlerContext.java // 省略代码 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 { ((ChannelInboundHandler) handler()).channelRead(this, msg); } catch (Throwable t) { invokeExceptionCaught(t); } } else { fireChannelRead(msg); } } /** * Makes best possible effort to detect if {@link ChannelHandler#handlerAdded(ChannelHandlerContext)} was called * yet. If not return {@code false} and if called or could not detect return {@code true}. * * If this method returns {@code false} we will not invoke the {@link ChannelHandler} but just forward the event. * This is needed as {@link DefaultChannelPipeline} may already put the {@link ChannelHandler} in the linked-list * but not called {@link ChannelHandler#handlerAdded(ChannelHandlerContext)}. */ private boolean invokeHandler() { // Store in local variable to reduce volatile reads. int handlerState = this.handlerState; return handlerState == ADD_COMPLETE || (!ordered && handlerState == ADD_PENDING); } // 省略代码