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

ゝ一纸荒年。 2022-05-16 12:43 193阅读 0赞

分析完inbound事件传播之后，outbound事件其实很简单，我们分以下几点解析：

1、何为outBound事件

2、outbound事件传播机制

3、ctx.channel().write()和ctx.write()的区别

### **一、何为outBound事件** ###

包含bind、connect、disconnect方法、close、deregister、read、write、flush等方法，这些方法更多的是主动向用户发起的操作。

（而inBound事件更多的是事件的触发，如register、readComplete、active，比较被动的）

**二、outbound事件传播机制**

首先看用户代码：

public final class Server1 {
    
        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 OutBoundHandlerA());
                                ch.pipeline().addLast(new OutBoundHandlerB());
                                ch.pipeline().addLast(new OutBoundHandlerC());
                            }
                        });
    
                ChannelFuture f = b.bind(8888).sync();
    
                f.channel().closeFuture().sync();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

添加了三个outbound事件处理器，按照A、B、C的顺序。

然后我们看各个outbound事件处理器，A和B一样，都是直接打印信息，并且传播写事件：

public class OutBoundHandlerA extends ChannelOutboundHandlerAdapter {
    
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            System.out.println("OutBoundHandlerA: " + msg);
            ctx.write(msg, promise);
        }
    }

我们看C：

public class OutBoundHandlerC extends ChannelOutboundHandlerAdapter {
    
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            System.out.println("OutBoundHandlerC: " + msg);
            ctx.write(msg, promise);
        }
        @Override
        public void handlerAdded(final ChannelHandlerContext ctx) {
            ctx.executor().schedule(() -> {
                ctx.channel().write("hello world");
                ctx.write("hello world");
            }, 3, TimeUnit.SECONDS);
        }
    }

C的write方法和AB一样，和他们不同的是C是事件传播的来源，只要它被添加进去，就发起写事件，也就是典型的outbound事件。我们从ctx.channel().write("hello world")开始进入：

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

继续：

@Override
        public final ChannelFuture write(Object msg) {
            return tail.write(msg);
        }

也就是它是从尾节点也就是TailContext开始传播的，一直进入：

private void write(Object msg, boolean flush, ChannelPromise promise) {
            AbstractChannelHandlerContext next = findContextOutbound();
            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 {
                AbstractWriteTask task;
                if (flush) {
                    task = WriteAndFlushTask.newInstance(next, m, promise);
                }  else {
                    task = WriteTask.newInstance(next, m, promise);
                }
                safeExecute(executor, task, promise, m);
            }
        }

它在这里执行findContextOutbound()查找下一个节点：

private AbstractChannelHandlerContext findContextOutbound() {
            AbstractChannelHandlerContext ctx = this;
            do {
                ctx = ctx.prev;
            } while (!ctx.outbound);
            return ctx;
        }

可以看到，它是从后面开始执行的，也就是：

tail-->C--->B-->A--->head。  
我们会到上一段，从这里开始next.invokeWriteAndFlush(m, promise);

private void invokeWriteAndFlush(Object msg, ChannelPromise promise) {
            if (invokeHandler()) {
                invokeWrite0(msg, promise);
                invokeFlush0();
            } else {
                writeAndFlush(msg, promise);
            }
        }

如果状态是对的，那就invokeWrite0()，否则，返回去继续循环。看invokeWrite0：

private void invokeWrite0(Object msg, ChannelPromise promise) {
            try {
                ((ChannelOutboundHandler) handler()).write(this, msg, promise);
            } catch (Throwable t) {
                notifyOutboundHandlerException(t, promise);
            }
        }

查找write的各自实现，如果是tail，刚刚介绍过，就是调用父类AbstractChannelHandlerContext的方法，直接找到next，然后传播。如果是C、B或者A：

System.out.println("OutBoundHandlerC: " + msg);
            ctx.write(msg, promise);

调用输出，然后继续是ctx.write(),这个方法和tail一样，继续调用父类构造方法寻找下一个节点，传播write事件。

最后如果是Head，就调用底层unsafe进行真正的写操作：

@Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                unsafe.write(msg, promise);
            }

### 三、ctx.channel().write()和ctx.write()的区别 ###

从刚刚分析源码中我们可以看到，ctx.channel().write()会从尾节点传递，经过所有的outboundhandler。

也就是如果我们注释掉ctx.write()，telnet 127.0.0.1 8888之后，结果会是：

OutBoundHandlerC: hello world
    OutBoundHandlerB: hello world
    OutBoundHandlerA: hello world

那么我们点进去ctx.write(),可以看到它最终调用到以下代码：

也就是，其实就是调用ctx父类AbstractChannelHandlerContext方法，传递给下一个，结果会是：

OutBoundHandlerB: hello world
    OutBoundHandlerA: hello world

不会调用tail（虽然调用tail也是执行调用下一个），也不会调用到自身handler的方法。

ctx.channel().write()从tail节点开始传播，ctx.write()从当前节点开始传播（不包括当前节点，因为它首先会找到当前节点的下一个节点在执行write操作）。

最后我们注意，context的write方法和handler的write方法意义是不同的。