netty源码之ChannelHandlerContext

谁借莪１个温暖的怀抱￠ 2022-12-27 09:17 168阅读 0赞

# 源码之ChannelHandlerContext #

在netty中，ChannelHandlerContext代表了一个ChannelHandler和ChannelPipeline之间的关系，Netty会把ChannelHandler包装进ChannelHandlerContext的实例DefaultChannelHandlerContext，然后把ChannelHandlerContext作为元素来组成链表，所以ChannelHandlerContext中一定有获得ChannelHandler和ChannelPipeline相关的方法。

ChannelHandler handler();
    ChannelPipeline pipeline();

ChannelHandlerContext主要功能是管理在同一ChannelPipeline中各个ChannelHandler的交互，所以同时ChannelHandlerContext也有很多触发事件传播相关的方法。

ChannelHandlerContext的具体实现就三个，都继承自AbstractChannelHandlerContext。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]

## AbstractChannelHandlerContext ##

AbstractChannelHandlerContext里面包含几乎所有与网络通信相关的方法，套路都一样，比如fireChannelRead方法，这是个inbound类型的方法，当调用ctx.fireChannelRead方法时，首先找到链表中下一个inbound 类型的handler，然后进行线程本地性判断，是则直接执行，不是则投入线程的队列异步执行。

fireChannelRead()方法的源码如下：

@Override
        public ChannelHandlerContext fireChannelRead(final Object msg) { 
            invokeChannelRead(findContextInbound(MASK_CHANNEL_READ), msg);
            return this;
        }
    
        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);
                    }
                });
            }
        }
    
        // 寻找双向链表中下一个入站context，也就是下一个入站handler
        private AbstractChannelHandlerContext findContextInbound(int mask) { 
            AbstractChannelHandlerContext ctx = this;
            EventExecutor currentExecutor = executor();
            do { 
                ctx = ctx.next;
            } while (skipContext(ctx, currentExecutor, mask, MASK_ONLY_INBOUND));
            return ctx;
        }

## HeadContext ##

HeadContext的继承关系如下：

final class HeadContext extends AbstractChannelHandlerContext
            implements ChannelOutboundHandler, ChannelInboundHandler {

可以看见，HeadContext既是ChannelHandlerContext，又是ChannelHandler，而且是同时处理inbound类型和outbound类型的ChannelHandler。

在具体的方法上，outbound类型事件的方法都是直接调用unsafe中的相关方法，而inbound类型事件的方法则会触发事件在ChannelPipeline中的传播。

## TailContext ##

TailContext的继承关系如下：

final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {

可以看见，TailContext既是ChannelHandlerContext，又是ChannelHandler，而且是处理inbound类型ChannelHandler。

在具体的方法上，TailContext里的方法基本上都是空实现，有具体代码实现的是：

protected void onUnhandledInboundUserEventTriggered(Object evt) { 
        // This may not be a configuration error and so don't log anything.
        // The event may be superfluous for the current pipeline configuration.
        ReferenceCountUtil.release(evt);
    }
    
    protected void onUnhandledInboundException(Throwable cause) { 
        try { 
            logger.warn(
                    "An exceptionCaught() event was fired, and it reached at the tail of the pipeline. " +
                            "It usually means the last handler in the pipeline did not handle the exception.",
                    cause);
        } finally { 
            ReferenceCountUtil.release(cause);
        }
    }
    
    protected void onUnhandledInboundMessage(Object msg) { 
        try { 
            logger.debug(
                    "Discarded inbound message {} that reached at the tail of the pipeline. " +
                            "Please check your pipeline configuration.", msg);
        } finally { 
            ReferenceCountUtil.release(msg);
        }
    }

这三个方法无一例外都是进行了相关资源的释放工作。

这里总结下平时在入站Handler中，释放资源的方式：

1.  继承SimpleChannelInboundHandler，实现channelRead0()方法。

我们来看下SimpleChannelInboundHandler的源码：

@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
        boolean release = true;
        try { 
            if (acceptInboundMessage(msg)) { 
                @SuppressWarnings("unchecked")
                I imsg = (I) msg;
                channelRead0(ctx, imsg); // 模板方法模式
            } else { 
                release = false;
                ctx.fireChannelRead(msg);
            }
        } finally { 
            if (autoRelease && release) { 
                ReferenceCountUtil.release(msg); // 释放资源
            }
        }
    }

1.  可以将资源向后传递给Tail进行释放。

## DefaultChannelHandlerContext ##

DefaultChannelHandlerContext这个里面的代码就非常简单了，大家自行查阅。

## ctx.write()、pipeline.write()、channel.write()的区别 ##

假设现在配置的Handler为：

p.addLast("1", new OutboundHandlerA());
    p.addLast("2", new InboundHandlerA());
    p.addLast("3", new InboundHandlerB()); // 在这个入站handler中分别调用三个write方法
    p.addLast("4", new OutboundHandlerB());

那么实际上出站Handler的个数为3个，分别如下：

HeadHandler --> OutboundHandlerA --> OutboundHandlerB

在入站InboundHandlerB中调用ctx.write()，所经过的出站Handler的个数为2个，分别如下：

OutboundHandlerA --> HeadHandler

在入站InboundHandlerB中调用pipeline.write()，所经过的出站Handler的个数为3个，分别如下：

OutboundHandlerB --> OutboundHandlerA -->HeadHandler

channel.write()与pipeline.write()的结果一致，实际上channel.write()底层调用了pipeline.write()。

下面是channel.write()的源码：

@Override
    public ChannelFuture writeAndFlush(Object msg) { 
        return pipeline.writeAndFlush(msg);
    }

pipeline.write()的源码：

// io.netty.channel.AbstractChannel#writeAndFlush(java.lang.Object)
    public final ChannelFuture writeAndFlush(Object msg) { 
        return tail.writeAndFlush(msg);
    }
    
    // io.netty.channel.AbstractChannelHandlerContext#writeAndFlush(java.lang.Object)
    public ChannelFuture writeAndFlush(Object msg) { 
        return writeAndFlush(msg, newPromise());
    }
    
    public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) { 
        write(msg, true, promise);
        return promise;
    }
    
    private void write(Object msg, boolean flush, ChannelPromise promise) { 
        ObjectUtil.checkNotNull(msg, "msg");
        try { 
            if (isNotValidPromise(promise, true)) { 
                ReferenceCountUtil.release(msg);
                // cancelled
                return;
            }
        } catch (RuntimeException e) { 
            ReferenceCountUtil.release(msg);
            throw e;
        }
    
        // 往前找最近的出站handler
        final AbstractChannelHandlerContext next = findContextOutbound(flush ?
                (MASK_WRITE | MASK_FLUSH) : MASK_WRITE);
        final Object m = pipeline.touch(msg, next);
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) { 
            if (flush) { 
                next.invokeWriteAndFlush(m, promise);
            } else { 
                next.invokeWrite(m, promise);
            }
        } else { 
            final WriteTask task = WriteTask.newInstance(next, m, promise, flush);
            if (!safeExecute(executor, task, promise, m, !flush)) { 
                // We failed to submit the WriteTask. We need to cancel it so we decrement the pending bytes
                // and put it back in the Recycler for re-use later.
                //
                // See https://github.com/netty/netty/issues/8343.
                task.cancel();
            }
        }
    }
    
    private AbstractChannelHandlerContext findContextOutbound(int mask) { 
        AbstractChannelHandlerContext ctx = this;
        EventExecutor currentExecutor = executor();
        do { 
            ctx = ctx.prev;
        } while (skipContext(ctx, currentExecutor, mask, MASK_ONLY_OUTBOUND));
        return ctx;
    }

从源码中可以看出pipeline.write()是从最后一个入站TailHandler往前开始遍历找出最近的出站Handler（这里是OutboundHandlerB），调用它的write方法，然后依次往前触发write方法。

下面是ctx.write()的源码，ctx.write()还是跟上面的一样，只不过上面的代码中this是TailHandler，所以会从TailHandler开始往前找最近的出站Handler（这里是OutboundHandlerB），而这里的this是InboundHandlerB，往前找会找到最近的出站Handler（这里是OutboundHandlerA）。

// io.netty.channel.AbstractChannelHandlerContext#writeAndFlush(java.lang.Object)
    public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) { 
        write(msg, true, promise);
        return promise;
    }

下面用一张图来总结下3个write的区别：

![在这里插入图片描述][20201210093553833.png_pic_center]

注意Tail是一个入站Handler，不是出站Handler。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/ae0ab9280c1b4fcba3f5a5fc321a7e5b.png
[20201210093553833.png_pic_center]: /images/20221120/49ab8aa79db84a0f875bfdea499a7eb4.png