netty源码阅读之pipeline之pipeline初始化

朴灿烈づ我的快乐病毒、 2022-05-16 01:36 286阅读 0赞

pipeline初始化我们将介绍以下几点：

1、pipeline在创建Channel的时候被创建

2、pipeline的节点的数据结构：ChannelHandlerContext

3、pipeline中的两大哨兵：head和tail

### 1、pipeline在创建Channel的时候被创建 ###

首先，我们记得，无论是创建服务端NioSocketChannel或者客户端SocketChannel的时候，最终都会调用到AbstractChannel的构造方法：

protected AbstractChannel(Channel parent) {
            this.parent = parent;
            id = newId();
            unsafe = newUnsafe();
            pipeline = newChannelPipeline();
        }

在这里创建pipeline，第一点就是这个意思。

那么从newChannelPipeline()进去(把自己channel当做参数传递进去)，一直进去：

protected DefaultChannelPipeline(Channel channel) {
            this.channel = ObjectUtil.checkNotNull(channel, "channel");
            succeededFuture = new SucceededChannelFuture(channel, null);
            voidPromise =  new VoidChannelPromise(channel, true);
    
            tail = new TailContext(this);
            head = new HeadContext(this);
    
            head.next = tail;
            tail.prev = head;
        }

两个哨兵就在这里：head和tail，head.next=tail，tail.prev=head，相信熟悉链表结构的人都会知道，他们构成一个双向链表。

head=HeadContext(this)这个，很熟悉吧。我们之前这篇文章《[netty源码阅读之服务器启动之端口绑定][netty]》的HeadContext，绑定端口。

让我们现在开始吧。

### 2、pipeline的节点的数据结构：ChannelHandlerContext ###

我们看head和tail，他们最终都继承自ChannelHandlerContext这个接口，不仅如此，pipeline的每一个节点都是这样的数据结构：

public interface ChannelHandlerContext extends AttributeMap, ChannelInboundInvoker, ChannelOutboundInvoker {}

AttributeMap这个很明显，就是保持了key和value的数据类型。

然后我们看ChannelInboundInvoker，

public interface ChannelInboundInvoker {
    
        ChannelInboundInvoker fireChannelRegistered();
    
        ChannelInboundInvoker fireChannelUnregistered();
    
        ChannelInboundInvoker fireChannelActive();
    
        ChannelInboundInvoker fireChannelInactive();
    
        ChannelInboundInvoker fireExceptionCaught(Throwable cause);
    
        ChannelInboundInvoker fireUserEventTriggered(Object event);
    
        ChannelInboundInvoker fireChannelRead(Object msg);
    
        ChannelInboundInvoker fireChannelWritabilityChanged();
    }

这里inbound事件，一般就是指读事件、注册事件或者active等；

然后看ChannelOutboundInvoker，

public interface ChannelOutboundInvoker {
    
        /**
         * Request to bind to the given {@link SocketAddress} and notify the {@link ChannelFuture} once the operation
         * completes, either because the operation was successful or because of an error.
         * <p>
         * This will result in having the
         * {@link ChannelOutboundHandler#bind(ChannelHandlerContext, SocketAddress, ChannelPromise)} method
         * called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
         * {@link Channel}.
         */
        ChannelFuture bind(SocketAddress localAddress);
    
        ChannelFuture connect(SocketAddress remoteAddress);
    
        ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress);
    
        ChannelFuture disconnect();
    
        ChannelFuture close();
    
        ChannelFuture deregister();
    
        ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise);
    
        ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise);
    
        ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise);
    
        ChannelFuture disconnect(ChannelPromise promise);
    
        ChannelFuture close(ChannelPromise promise);
    
        ChannelFuture deregister(ChannelPromise promise);
    
        ChannelOutboundInvoker read();
    
        ChannelFuture write(Object msg);
    
        ChannelFuture write(Object msg, ChannelPromise promise);
    
        ChannelOutboundInvoker flush();
    
        ChannelFuture writeAndFlush(Object msg, ChannelPromise promise);
    
        ChannelFuture writeAndFlush(Object msg);
    
        ChannelPromise newPromise();
    
        ChannelProgressivePromise newProgressivePromise();
    
        ChannelFuture newSucceededFuture();
    
        ChannelFuture newFailedFuture(Throwable cause);
    
        ChannelPromise voidPromise();
    }

里面的这些方法，主要靠这个ChannelOutboundInvoker处理（主要是传播写事件吧）。

最后看ChannelHandlerContext自身：

public interface ChannelHandlerContext extends AttributeMap, ChannelInboundInvoker, ChannelOutboundInvoker {
    
        /**
         * Return the {@link Channel} which is bound to the {@link ChannelHandlerContext}.
         */
        Channel channel();
    
        /**
         * Returns the {@link EventExecutor} which is used to execute an arbitrary task.
         */
        EventExecutor executor();
    
        /**
         * The unique name of the {@link ChannelHandlerContext}.The name was used when then {@link ChannelHandler}
         * was added to the {@link ChannelPipeline}. This name can also be used to access the registered
         * {@link ChannelHandler} from the {@link ChannelPipeline}.
         */
        String name();
    
        /**
         * The {@link ChannelHandler} that is bound this {@link ChannelHandlerContext}.
         */
        ChannelHandler handler();
    
        /**
         * Return {@code true} if the {@link ChannelHandler} which belongs to this context was removed
         * from the {@link ChannelPipeline}. Note that this method is only meant to be called from with in the
         * {@link EventLoop}.
         */
        boolean isRemoved();
        ...
        /**
         * Return the assigned {@link ByteBufAllocator} which will be used to allocate {@link ByteBuf}s.
         */
        ByteBufAllocator alloc();
        ...
    }

自己的一些方法主要就是上面的几个方法。channel说明pipeline的每个节点都要了解自己是哪一个channel的；executor说明每个节点都需要知道现在是和哪一个NioEventLoop绑定在一起的；真正在做事情的就是这个handler；还有一个ByteBufAllocator，内存的分配器，这就说明在这个节点这里，如果有数据读写要分配ByteBuf的时候，需要使用哪一个内存分配器去分配。

AbstractChannelHandlerContext是ChannelHandlerContext这个类的实现，实现了大部分的功能，里面主要有个next和prev的指针，把这些handler连起来。

### 3、pipeline中的两大哨兵 ###

TailContext先看类继承关系图：

![70][]

TailContext它的源码在DefaultChannelPipeline里面：

// A special catch-all handler that handles both bytes and messages.
        final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {
    
            TailContext(DefaultChannelPipeline pipeline) {
                super(pipeline, null, TAIL_NAME, true, false);
                setAddComplete();
            }
    
            @Override
            public ChannelHandler handler() {
                return this;
            }
    
            @Override
            public void channelRegistered(ChannelHandlerContext ctx) throws Exception { }
    
            @Override
            public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { }
    
            @Override
            public void channelActive(ChannelHandlerContext ctx) throws Exception { }
    
            @Override
            public void channelInactive(ChannelHandlerContext ctx) throws Exception { }
    
            @Override
            public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { }
    
            @Override
            public void handlerAdded(ChannelHandlerContext ctx) throws Exception { }
    
            @Override
            public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { }
    
            @Override
            public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                // 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);
            }
    
            @Override
            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                onUnhandledInboundException(cause);
            }
    
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                onUnhandledInboundMessage(msg);
            }
    
            @Override
            public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { }
        }

这个类比较简单，可以看到，它既处理bytes事件的，也处理messages事件的，也就是，既处理服务器的请求也处理客户端的请求。构造函数，inbound为true，outbound为false（说明他是一个inbound处理器）。handler就是它自己this（说明他不仅是一个节点而且业务处理器也是它自己）。总体来说，只处理以下两个事件：

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

发生了一些异常（或者读事件）了，就要处理，但是最终处理异常（或者读事件）信息的时候，只是把它打印出来（如果是读事件的话，会把message看看是否能释放，可以释放就释放了）：

/**
         * Called once a {@link Throwable} hit the end of the {@link ChannelPipeline} without been handled by the user
         * in {@link ChannelHandler#exceptionCaught(ChannelHandlerContext, Throwable)}.
         */
        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);
            }
        }
    
        /**
         * Called once a message hit the end of the {@link ChannelPipeline} without been handled by the user
         * in {@link ChannelInboundHandler#channelRead(ChannelHandlerContext, Object)}. This method is responsible
         * to call {@link ReferenceCountUtil#release(Object)} on the given msg at some point.
         */
        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);
            }
        }

可以看到tail主要做一些收尾的信息，如果你有一些事件没处理，它就帮你处理了。

然后看HeadContext，这个类比TailContext“高级”一点：

![70 1][]

首先它多实现了一个ChannelOutboundHandler接口，源码里面它就多做了很多事，首先就是大部分接口都实现了：

final class HeadContext extends AbstractChannelHandlerContext
                implements ChannelOutboundHandler, ChannelInboundHandler {
    
            private final Unsafe unsafe;
    
            HeadContext(DefaultChannelPipeline pipeline) {
                super(pipeline, null, HEAD_NAME, false, true);
                unsafe = pipeline.channel().unsafe();
                setAddComplete();
            }
    
            @Override
            public ChannelHandler handler() {
                return this;
            }
    
            @Override
            public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
                // NOOP
            }
    
            @Override
            public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
                // NOOP
            }
    
            @Override
            public void bind(
                    ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise)
                    throws Exception {
                unsafe.bind(localAddress, promise);
            }
    
            @Override
            public void connect(
                    ChannelHandlerContext ctx,
                    SocketAddress remoteAddress, SocketAddress localAddress,
                    ChannelPromise promise) throws Exception {
                unsafe.connect(remoteAddress, localAddress, promise);
            }
    
            @Override
            public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
                unsafe.disconnect(promise);
            }
    
            @Override
            public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
                unsafe.close(promise);
            }
    
            @Override
            public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
                unsafe.deregister(promise);
            }
    
            @Override
            public void read(ChannelHandlerContext ctx) {
                unsafe.beginRead();
            }
    
            @Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                unsafe.write(msg, promise);
            }
    
            @Override
            public void flush(ChannelHandlerContext ctx) throws Exception {
                unsafe.flush();
            }
    
            @Override
            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                ctx.fireExceptionCaught(cause);
            }
    
            @Override
            public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
                invokeHandlerAddedIfNeeded();
                ctx.fireChannelRegistered();
            }
    
            @Override
            public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
                ctx.fireChannelUnregistered();
    
                // Remove all handlers sequentially if channel is closed and unregistered.
                if (!channel.isOpen()) {
                    destroy();
                }
            }
    
            @Override
            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                ctx.fireChannelActive();
    
                readIfIsAutoRead();
            }
    
            @Override
            public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                ctx.fireChannelInactive();
            }
    
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                ctx.fireChannelRead(msg);
            }
    
            @Override
            public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
                ctx.fireChannelReadComplete();
    
                readIfIsAutoRead();
            }
    
            private void readIfIsAutoRead() {
                if (channel.config().isAutoRead()) {
                    channel.read();
                }
            }
    
            @Override
            public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                ctx.fireUserEventTriggered(evt);
            }
    
            @Override
            public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
                ctx.fireChannelWritabilityChanged();
            }
        }

看源码里面的这段代码是不是很熟悉啊：

@Override
            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                ctx.fireChannelActive();
    
                readIfIsAutoRead();
            }

然后还多了一个成员变量Unsafe，如果是客户端的有客户端的实现（NioByteUnsafe），服务端的有服务端的实现（NioMessageUnsafe），用于实现底层数据的读写。

注意到它有的构造函数inbound是false，outbound是true，和TailContext相反（当时和我们直觉相反）。

然后它也实现了绑定连接等操作，绑定的话我们在服务端绑定端口的[这篇文章][netty]有做了介绍：

@Override
            public void bind(
                    ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise)
                    throws Exception {
                unsafe.bind(localAddress, promise);
            }
    
            @Override
            public void connect(
                    ChannelHandlerContext ctx,
                    SocketAddress remoteAddress, SocketAddress localAddress,
                    ChannelPromise promise) throws Exception {
                unsafe.connect(remoteAddress, localAddress, promise);
            }
    
            @Override
            public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
                unsafe.disconnect(promise);
            }

HeadContext和TailContext区别很大，HeadContext主要是把事件原模原样地往下传播，传播事件的时候，都会从head开始；在一些进行读写操作的时候，都会委托到head的unsafe这里操作，也就是HeadContext负责channel具体协议的实现。

而TailContext负责终止事件和异常传播的作用。

[netty]: https://blog.csdn.net/fst438060684/article/details/81431560
[70]: /images/20220516/c001a2e710d642fab4c3806d2eac398e.png
[70 1]: /images/20220516/ddf9a43526db48f59d50033f9e6a41de.png