netty源码阅读之pipeline之inBound事件传播

╰半橙微兮° 2022-05-16 10:44 246阅读 0赞

我们通过分析以下几点，分析inBound事件的传播。

1、何为inBound事件以及ChannelInBoundHandler

2、ChannelRead事件的传播

3、SimpleInBoundHandler处理器

首先我们看一个类图：

![70][]

我们可以知道，ChannelInBoundHandler和ChannelOutboundHandlerAdapter都来源于ChannelHandler，基类ChannelHandlerAdapter实现了基础功能，通过ChannelInboundHandler和ChannelOutboundHandler接口实现他们各自的功能。关于这个类图怎么画，[这里][Link 1]有一篇很好的介绍。

### 一、何为inBound事件以及ChannelInBoundHandler ###

先看ChannelHandler怎么源码：

public interface ChannelHandler {
 
 /**
 * Gets called after the {@link ChannelHandler} was added to the actual context and it's ready to handle events.
 */
 void handlerAdded(ChannelHandlerContext ctx) throws Exception;
 
 /**
 * Gets called after the {@link ChannelHandler} was removed from the actual context and it doesn't handle events
 * anymore.
 */
 void handlerRemoved(ChannelHandlerContext ctx) throws Exception;
 
 /**
 * Gets called if a {@link Throwable} was thrown.
 *
 * @deprecated is part of {@link ChannelInboundHandler}
 */
 @Deprecated
 void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception;
 
 /**
 * Indicates that the same instance of the annotated {@link ChannelHandler}
 * can be added to one or more {@link ChannelPipeline}s multiple times
 * without a race condition.
 * 
 * If this annotation is not specified, you have to create a new handler
 * instance every time you add it to a pipeline because it has unshared
 * state such as member variables.
 * 
 * This annotation is provided for documentation purpose, just like
 * <a href="http://www.javaconcurrencyinpractice.com/annotations/doc/">the JCIP annotations</a>.
 */
 @Inherited
 @Documented
 @Target(ElementType.TYPE)
 @Retention(RetentionPolicy.RUNTIME)
 @interface Sharable {
 // no value
 }
 }

接口就是定义规范，这里定义了最基本的，add，remove，exception发生各自的回调，最后还有一个注解，是否共享的，在[这篇文章][Link 2]里面介绍过。

ChannelHandlerAdapter就是实现了一小部分ChannelHandler的功能，自行查看。

然后我们看ChannelInboundHandler：

/**
     * {@link ChannelHandler} which adds callbacks for state changes. This allows the user
     * to hook in to state changes easily.
     */
    public interface ChannelInboundHandler extends ChannelHandler {
    
        /**
         * The {@link Channel} of the {@link ChannelHandlerContext} was registered with its {@link EventLoop}
         */
        void channelRegistered(ChannelHandlerContext ctx) throws Exception;
    
        /**
         * The {@link Channel} of the {@link ChannelHandlerContext} was unregistered from its {@link EventLoop}
         */
        void channelUnregistered(ChannelHandlerContext ctx) throws Exception;
    
        /**
         * The {@link Channel} of the {@link ChannelHandlerContext} is now active
         */
        void channelActive(ChannelHandlerContext ctx) throws Exception;
    
        /**
         * The {@link Channel} of the {@link ChannelHandlerContext} was registered is now inactive and reached its
         * end of lifetime.
         */
        void channelInactive(ChannelHandlerContext ctx) throws Exception;
    
        /**
         * Invoked when the current {@link Channel} has read a message from the peer.
         */
        void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception;
    
        /**
         * Invoked when the last message read by the current read operation has been consumed by
         * {@link #channelRead(ChannelHandlerContext, Object)}.  If {@link ChannelOption#AUTO_READ} is off, no further
         * attempt to read an inbound data from the current {@link Channel} will be made until
         * {@link ChannelHandlerContext#read()} is called.
         */
        void channelReadComplete(ChannelHandlerContext ctx) throws Exception;
    
        /**
         * Gets called if an user event was triggered.
         */
        void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception;
    
        /**
         * Gets called once the writable state of a {@link Channel} changed. You can check the state with
         * {@link Channel#isWritable()}.
         */
        void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception;
    
        /**
         * Gets called if a {@link Throwable} was thrown.
         */
        @Override
        @SuppressWarnings("deprecation")
        void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception;
    }

首先我们由介绍开始查看：

@link ChannelHandler} which adds callbacks for state changes. This allows the user
    to hook in to state changes easily.

状态的变化，添加回调。然后我们对比ChannelOutboundHandler源码的介绍：

@link ChannelHandler} which will get notified for IO-outbound-operations.

io操作后的通知。

从status changes和operations中我们细细琢磨，可以知道inbound事件是比较被动的，而outbound事件是比较主动的，因为它是一个“operations”。

我们继续看ChannelInboundHandler实现的方法：

channelRegistered

channelUnregistered

channelActive

channelInactive

channelRead

channelReadComplete

userEventTriggered

channelWritabilityChanged

exceptionCaught

这下就大概明白了。

### 二、ChannelRead事件的传播 ###

我们以ChannelRead事件为例来分析inBound事件。

我们有一段客户端代码：

public final class Server {
 
 public static void main(String[] args) throws Exception {
 EventLoopGroup bossGroup = new NioEventLoopGroup(1);
 EventLoopGroup workerGroup = new NioEventLoopGroup();
 
 try {
 ServerBootstrap b = new ServerBootstrap();
 b.group(bossGroup, workerGroup)
 .channel(NioServerSocketChannel.class)
 .childOption(ChannelOption.TCP_NODELAY, true)
 .childAttr(AttributeKey.newInstance("childAttr"), "childAttrValue")
 .childHandler(new ChannelInitializer<SocketChannel>() {
 @Override
 public void initChannel(SocketChannel ch) {
 ch.pipeline().addLast(new InBoundHandlerA());
 ch.pipeline().addLast(new InBoundHandlerB());
 ch.pipeline().addLast(new InBoundHandlerC());
 }
 });
 
 ChannelFuture f = b.bind(8888).sync();
 
 f.channel().closeFuture().sync();
 } finally {
 bossGroup.shutdownGracefully();
 workerGroup.shutdownGracefully();
 }
 }
 }

每个InBoundHandlerX()，X代表A或者B或者C，里面都是这样实现的：

public class InBoundHandlerA extends ChannelInboundHandlerAdapter {
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println("InBoundHandlerA: " + msg);
            ctx.fireChannelRead(msg);
        }
    }

以A为例子，其他不赘述。

运行之后，我们telnet一下本地端口8888，能得到一下结果：

InBoundHandlerA: hello world
    InBoundHandlerB: hello world
    InBoundHandlerC: hello world

这说明，InBound时间的传播，是按照添加顺序来的。

也就是Head-->A-->B-->C--->Tail

那么我们进入HeadContext的ChannelRead，有新连接进来之后先调用这个方法：

@Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                ctx.fireChannelRead(msg);
            }

这里没做什么，直接传递事件下去，怎么传递下去？点击查看fireChannelRead的源码：

@Override
        public ChannelHandlerContext fireChannelRead(final Object msg) {
            invokeChannelRead(findContextInbound(), msg);
            return this;
        }

这里有一个findContextInbound()，也就是找下一个inboundhandler，然后通过invokeChannelRead传递下去，查看findContextInbound()源码：

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

这个源码说明了，inbound是按照添加的顺序执行的，因为每一个都找next指针的节点。这段逻辑是什么呢？如果是inbound处理器，ctx=ctx.next的ctx就可以返回了，也就是，如果是inbound处理器，就返回。我们怎么知道是inbound呢，就通过inbound标识。回到[这篇][Link 2]添加handler的文章，我们将inboundHanlder包装到context里面这段代码：

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

继续：

DefaultChannelHandlerContext(
                DefaultChannelPipeline pipeline, EventExecutor executor, String name, ChannelHandler handler) {
            super(pipeline, executor, name, isInbound(handler), isOutbound(handler));
            if (handler == null) {
                throw new NullPointerException("handler");
            }
            this.handler = handler;
        }

先看isInbound怎么实现，太简单：

private static boolean isInbound(ChannelHandler handler) {
            return handler instanceof ChannelInboundHandler;
        }

点击去super方法：

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;
            // Its ordered if its driven by the EventLoop or the given Executor is an instanceof OrderedEventExecutor.
            ordered = executor == null || executor instanceof OrderedEventExecutor;
        }

答案就在这里。

另外，如果是HeadContext或者TailContext的时候，我们自己设置inbound和outbound为true或者false。

[这里][Link 3]pipeline两大哨兵的时候有介绍。

返回之前，继续查看invokeChannelRead源码：

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);
                    }
                });
            }
        }

在这里是inEventLoop，所以继续进入，next.invokeChannelRead(m)，到达这里：

private void invokeChannelRead(Object msg) {
            if (invokeHandler()) {
                try {
                    ((ChannelInboundHandler) handler()).channelRead(this, msg);
                } catch (Throwable t) {
                    notifyHandlerException(t);
                }
            } else {
                fireChannelRead(msg);
            }
        }

找到InBoundHandlerA的实现，我们可以发现进到它的这个方法了：

@Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println("InBoundHandlerA: " + msg);
            ctx.fireChannelRead(msg);
        }

可以通过单步调试的方式验证。因为A、B、C、Head都是包装自AbstractChannelHandlerContext的，所以传播的源码是一样的。至于行为A、B、C，那就是打印A、B、C各自的ChannelRead用户行为。我们看一下tail的ChannelRead事件：

@Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                onUnhandledInboundMessage(msg);
            }

继续：

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);
            }
        }

其实我们之前[这里][Link 3]也讲过，它就是不处理，打印日志提示你没有处理这个事件，并且释放资源。每次用户代码处理完之后，如果不传播到tail节点，都要自己释放资源，那么有没有自动释放资源的办法的?答案当然有，继承以下类：

### 三、SimpleInBoundHandler处理器 ###

先看源码，一睹为快：

public abstract class SimpleChannelInboundHandler extends ChannelInboundHandlerAdapter {
 
 private final TypeParameterMatcher matcher;
 private final boolean autoRelease;
 
 ...
 
 @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);
 }
 }
 }
 
 /**
 * Please keep in mind that this method will be renamed to
 * {@code messageReceived(ChannelHandlerContext, I)} in 5.0.
 *
 * Is called for each message of type {@link I}.
 *
 * @param ctx the {@link ChannelHandlerContext} which this {@link SimpleChannelInboundHandler}
 * belongs to
 * @param msg the message to handle
 * @throws Exception is thrown if an error occurred
 */
 protected abstract void channelRead0(ChannelHandlerContext ctx, I msg) throws Exception;
 }

看channelRead方法，通过channelRead0()执行具体的逻辑，最后有一个释放的过程。所以用户代码继承他，是覆盖channelRead0()方法执行具体的逻辑，我们看一个用户代码：

public class AuthHandler extends SimpleChannelInboundHandler<ByteBuf> {
 @Override
 protected void channelRead0(ChannelHandlerContext ctx, ByteBuf password) throws Exception {
 if (paas(password)) {
 ctx.pipeline().remove(this);
 } else {
 ctx.close();
 }
 }
 
 private boolean paas(ByteBuf password) {
 return false;
 }
 }

但是这里有个ByteBuf，这个用来做什么的，具体的实现，我们后面会去讲解。具体到这个例子，用在什么地方呢？

它是以一个泛型的方式添加进去的，我们可以大致理解为只有这个类型的数据才处理，因为在channelRead里面acceptInboundMessage(msg)，这个就是返回类型是否相同：

public boolean acceptInboundMessage(Object msg) throws Exception {
            return matcher.match(msg);
        }

private static final class ReflectiveMatcher extends TypeParameterMatcher {
 private final Class<?> type;
 
 ReflectiveMatcher(Class<?> type) {
 this.type = type;
 }
 
 @Override
 public boolean match(Object msg) {
 return type.isInstance(msg);
 }
 }

而上面的type就是创建matcher的时候传进去的，也就是泛型ByteBuf：

protected SimpleChannelInboundHandler(Class<? extends I> inboundMessageType, boolean autoRelease) {
 matcher = TypeParameterMatcher.get(inboundMessageType);
 this.autoRelease = autoRelease;
 }

可以继续看源码深究。

[70]: /images/20220516/cef6a824f9c1403b9a3b0f993a48ba68.png
[Link 1]: https://www.cnblogs.com/imsoft/p/6972612.html
[Link 2]: https://blog.csdn.net/fst438060684/article/details/81703748
[Link 3]: https://blog.csdn.net/fst438060684/article/details/81661226