Netty之Pipeline的初始化

深碍√TFBOYSˉ_ 2023-06-29 10:45 7阅读 0赞

### Pipeline的初始化 ###

Pipeline的初始化逻辑是在AbstractChannel的构造函数中完成的，在其子类NioServerSocketChannel实例化时触发。

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

**newChannelPipeline**

方法的主要逻辑就是将当前channel作为参数传入并缓存，然后创建头尾节点，构建一个双向链表。

protected DefaultChannelPipeline newChannelPipeline() {
            return new DefaultChannelPipeline(this);
        }
    
        protected DefaultChannelPipeline(Channel 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;
        }

--------------------

### Pipeline节点数据结构ChannelHandlerContext ###

TailContext与HeadContext都继承于AbstractChannelHandlerContext，并实现了ChannelHandlerContext接口。

/**
     * AttributeMap: 存储一些自定义属性
     * ChannelInboundInvoker: 主要定义读事件
     * ChannelOutboundInvoker: 主要定义写事件
     */
    public interface ChannelHandlerContext extends AttributeMap, ChannelInboundInvoker, ChannelOutboundInvoker {
    
        // 当前节点所属的channel
        Channel channel();
    
        // 当前节点的NioEventLoop
        EventExecutor executor();
    
        // 业务处理器名称
        String name();
    
        // 业务处理器
        ChannelHandler handler();
    
        // 当前节点是否被移除
        boolean isRemoved();
    
        // 一些列inbound与outbound事件
        @Override
        ChannelHandlerContext fireChannelRegistered();
    
        @Override
        ChannelHandlerContext fireChannelUnregistered();
    
        @Override
        ChannelHandlerContext fireChannelActive();
    
        @Override
        ChannelHandlerContext fireChannelInactive();
    
        @Override
        ChannelHandlerContext fireExceptionCaught(Throwable cause);
    
        @Override
        ChannelHandlerContext fireUserEventTriggered(Object evt);
    
        @Override
        ChannelHandlerContext fireChannelRead(Object msg);
    
        @Override
        ChannelHandlerContext fireChannelReadComplete();
    
        @Override
        ChannelHandlerContext fireChannelWritabilityChanged();
    
        @Override
        ChannelHandlerContext read();
    
        @Override
        ChannelHandlerContext flush();
    
        // 当前节点的pipeline
        ChannelPipeline pipeline();
    
        // ByteBuffer分配器
        ByteBufAllocator alloc();
    
        // 添加属性
        @Deprecated
        @Override
        <T> Attribute<T> attr(AttributeKey<T> key);
    
        // 判断属性是否存在
        @Deprecated
        @Override
        <T> boolean hasAttr(AttributeKey<T> key);
    }

--------------------

### Pipeline的两大哨兵：head和tail ###

**TailContext**

tail节点传播inbound事件，主要做一些收尾工作，如记录异常未处理、消息未处理的警告日志。

final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {
    
            TailContext(DefaultChannelPipeline pipeline) {
                // inbound:true
                // outbound:false
                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 { }
        }

**HeadContext**

head节点传播outbound事件，用于传播读写事件，具体的读写委托unsafe对象实现。

final class HeadContext extends AbstractChannelHandlerContext
                implements ChannelOutboundHandler, ChannelInboundHandler {
    
            private final Unsafe unsafe;
    
            HeadContext(DefaultChannelPipeline pipeline) {
                // inbound:false
                // outbound:true
                super(pipeline, null, HEAD_NAME, false, true);
                // unsafe对象用于实现具体的读写操作
                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();
            }
        }