netty5笔记-总体流程分析3-ChannelHandlerContext

上面一篇文章介绍了ChannelPipeline，它维护了一个有序的ChannelHandler列表，但并非是直接关联，而是通过维护ChannelHandlerContext进行关联。ChannelPipeline的每一个节点都是一个ChannelHandlerContext实例，不存在线程安全问题，而对应的ChannelHandler如果未标记Sharable也是新建的实例，也不存在线程安全问题。对于Sharable的ChannelHandler则由实现者去处理线程安全问题。可以说ChannelHandlerContext起到了一个桥梁的作用，它可以方便的为ChannelHandler传递数据到位于同一ChannelPipeline前面或者后面的ChannelHander，同时它也可以为当前ChannelHandler存储有状态的信息。 总结一下，一个ChannelHandlerContext只能对应一个ChannelHander，只对应一个Channel，而一个ChannelHander则可以对应多个ChannelHandlerContext。在前一篇中我们看到ChannelPipeline中定义了一些列outbound、inbound事件，对应的方法都会调用ChannelHandlerContext中的对应方法，由于方法名一致，这里不再介绍了，看看ChannelHanderContext中特有的几个方法：

 

方法名 说明

channel	获取与Context关联的Channel, 一个Context对应一个Channel,而一个Channel则对应了多个Context
executor	用于执行实例中产生的Promise中的listenrer
invoker	通过该对象调用handler上的各个方法，通过它的调用可以保证线程安全
handler	实际的事件处理者

上面的invoker可以保证线程安全，它是如何做的呢？ 我们来看看其中一个invoker的部分实现：

 

    
    public void invokeChannelInactive(final ChannelHandlerContext ctx) {
        // 当前调用如果是在EventLoop中则直接调用xxxNow （inEventLoop保证线程安全，见前面的EventLoop分析）
       if (executor.inEventLoop()) {
            invokeChannelInactiveNow(ctx);
        } else {
            // 如果不在EventLoop中则调用execute方法，此处execute方法将此任务添加到EventLoop的任务队列中，等待后续执行
            executor.execute(new OneTimeTask() {
                @Override
                public void run() {
                    invokeChannelInactiveNow(ctx);
                }
            });
        }
    }

    // 下面的几个方法来源于ChannelHanderInvokerUtil
    public static void invokeChannelInactiveNow(final ChannelHandlerContext ctx) {
 try {
            // 触发handler()的对应方法
 ctx.handler().channelInactive(ctx);
 } catch (Throwable t) {
            // 出现异常则调用异常处理逻辑
 notifyHandlerException(ctx, t);
 }
 }

    private static void notifyHandlerException(ChannelHandlerContext ctx, Throwable cause) {
        // 异常处理逻辑先判断该异常是否来自于另一个异常，如果是则不再向下传递，避免递归调用
 if (inExceptionCaught(cause)) {
 if (logger.isWarnEnabled()) {
 logger.warn(
 "An exception was thrown by a user handler " +
 "while handling an exceptionCaught event", cause);
 }
 return;
 }

        // 调用实际的异常处理方法
 invokeExceptionCaughtNow(ctx, cause);
 }

    public static void invokeExceptionCaughtNow(final ChannelHandlerContext ctx, final Throwable cause) {
        // 调用handler的异常处理逻辑，如果异常处理出现异常，不再向下传递
 try {
 ctx.handler().exceptionCaught(ctx, cause);
 } catch (Throwable t) {
 if (logger.isWarnEnabled()) {
 logger.warn("An exception was thrown by a user handler's exceptionCaught() method:", t);
 logger.warn(".. and the cause of the exceptionCaught() was:", cause);
 }
 }
 }

    private static boolean inExceptionCaught(Throwable cause) {
 do {
 StackTraceElement[] trace = cause.getStackTrace();
 if (trace != null) {
 for (StackTraceElement t : trace) {
 if (t == null) {
 break;
 }
 if ("exceptionCaught".equals(t.getMethodName())) {
 return true;
 }
 }
 }

 cause = cause.getCause();
 } while (cause != null);

 return false;
 }

从上面几个方法可以看到ChannelHandlerInvoker通过判断当前方法是否在执行线程中来决定是马上执行方法还是放到队列中等待后面执行。如果马上执行，则调用handler中的对应方法，并加上对应的异常处理逻辑，这也是netty健壮性的一个保证。也就是说context方法的调用，不管是在IO事件循环中调用还是在用户线程中调用，都是线程安全的。

了解Invoker的作用，我们再回到Context, 分别取了inbound和outbound的两个方法，调用过程很简单，首先是findContextXXXBound，然后通过它调用invoker, invoker调用最终的handler方法。findContextOutbound和findContextInbound查找的方向正好相反，前一个是从当前context往前找，后一个是从当前context往后找。这个也印证了ChannelPipeline中的那幅图。 要注意的是如何判断一个context是否可以处理outbound/inbound事件， 从代码实现可以看到使用的是位操作来进行判断

 

     public ChannelHandlerContext fireChannelActive() {
        AbstractChannelHandlerContext next = findContextInbound();
        next.invoker().invokeChannelActive(next);
        return this;
    }

    public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
 AbstractChannelHandlerContext next = findContextOutbound();
 next.invoker().invokeBind(next, localAddress, promise);
 return promise;
 }
    // 查找下一个Inbound
    private AbstractChannelHandlerContext findContextInbound() {
 AbstractChannelHandlerContext ctx = this;
 do {
 ctx = ctx.next;
 } while ((ctx.skipFlags & MASKGROUP_INBOUND) == MASKGROUP_INBOUND);
 return ctx;
 }
    // 查找下一个Outbound
 private AbstractChannelHandlerContext findContextOutbound() {
 AbstractChannelHandlerContext ctx = this;
 do {
 ctx = ctx.prev;
 } while ((ctx.skipFlags & MASKGROUP_OUTBOUND) == MASKGROUP_OUTBOUND);
 return ctx;
 }

可以看到MASKGROUP_INBOUND和MASKGROUP_OUTBOUND是通过按位或的方式来计算得到的，每一个MASK_XXX都是2的N次方，这样我们就可以很简单的通过按位与知道是否满足某个标记。

 

private static final int MASKGROUP_INBOUND = MASK_EXCEPTION_CAUGHT |
            MASK_CHANNEL_REGISTERED |
            MASK_CHANNEL_UNREGISTERED |
            MASK_CHANNEL_ACTIVE |
            MASK_CHANNEL_INACTIVE |
            MASK_CHANNEL_READ |
            MASK_CHANNEL_READ_COMPLETE |
            MASK_CHANNEL_WRITABILITY_CHANGED |
            MASK_USER_EVENT_TRIGGERED;

    private static final int MASKGROUP_OUTBOUND = MASK_BIND |
            MASK_CONNECT |
            MASK_DISCONNECT |
            MASK_CLOSE |
            MASK_DEREGISTER |
            MASK_READ |
            MASK_WRITE |
            MASK_FLUSH;

skipFlags是通过反射获取注解的方式来计算的，我们知道反射的性能是非常低的，因此这里使用了WeakHashMap的方式对结果进行缓存。

 

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

 static int skipFlags(ChannelHandler handler) {
        // 首先从缓存中获取如果获取不到再计算
 WeakHashMap<Class<?>, Integer> cache = skipFlagsCache.get();
 Class<? extends ChannelHandler> handlerType = handler.getClass();
 int flagsVal;
 Integer flags = cache.get(handlerType);
 if (flags != null) {
 flagsVal = flags;
 } else {
 flagsVal = skipFlags0(handlerType);
 cache.put(handlerType, Integer.valueOf(flagsVal));
 }

 return flagsVal;
 }

    // 计算的方式就是依次判断对应的handler是否实现了某个方法，如果实现了，则通过位或的方式加入到flag。而是否实现的标准就是对应方法是否有Ship的注解，
    // 如果有则表示未实现对应方法（默认的ChannelHandlerAdapter就是实现的所有ChannelHandler的方法都添加了Skip注解。要注意的是由于findContextXXBound的时候
    // 主要handler实现了inbound/outbound中的任意一个方法，那他的对应方法都是会执行的。 即如果某个handler只实现了channelRead方法，
    // 该handler的其他inbound方法依然会被调用，为不会被忽略。
 static int skipFlags0(Class<? extends ChannelHandler> handlerType) {
 int flags = 0;
 try {
 if (isSkippable(handlerType, "handlerAdded")) {
 flags |= MASK_HANDLER_ADDED;
 }
 if (isSkippable(handlerType, "handlerRemoved")) {
 flags |= MASK_HANDLER_REMOVED;
 }
 ....中间都是类似的代码，就不贴了....
 if (isSkippable(handlerType, "write", Object.class, ChannelPromise.class)) {
 flags |= MASK_WRITE;
 }
 if (isSkippable(handlerType, "flush")) {
 flags |= MASK_FLUSH;
 }
 } catch (Exception e) {
 // Should never reach here.
 PlatformDependent.throwException(e);
 }

 return flags;
 }

 private static boolean isSkippable(
 Class<?> handlerType, String methodName, Class<?>... paramTypes) throws Exception {

 Class[] newParamTypes = new Class[paramTypes.length + 1];
 newParamTypes[0] = ChannelHandlerContext.class;
 System.arraycopy(paramTypes, 0, newParamTypes, 1, paramTypes.length);

 return handlerType.getMethod(methodName, newParamTypes).isAnnotationPresent(Skip.class);
 }

其他的方法相对来说也比较简单且重复，我们就不必讲这么细了。

总结一下，每一个请求都会创建ChannelPipeline,ChannelPipeline内部包含一个有ChannelHandlerContext组成的双向链表，context通过handler的skipFlags来查找下一个inbound/outbound，而具体的调用则通过invoker来完成，invoker保证了调用的线程安全，最终线程安全的调用了handler中的具体方法。

context主要负责inbound/outbound事件的传递，由于context不会共用，还可以安全的存储自定义信息（见AttributeMap）。