深入淺出Netty：ChannelPipeline

阿新 • • 發佈：2019-01-14

每個channel內部都會持有一個ChannelPipeline物件pipeline。pipeline預設實現DefaultChannelPipeline內部維護了一個DefaultChannelHandlerContext連結串列。

2184951-beacd91367f1f4eb

當channel完成register、active、read等操作時，會觸發pipeline的相應方法。

1、當channel註冊到selector時，觸發pipeline的fireChannelRegistered方法。
2、當channel的socket繫結完成時，觸發pipeline的fireChannelActive方法。
3、當有客戶端請求時，觸發pipeline的fireChannelRead方法。
4、當本次客戶端請求，pipeline執行完fireChannelRead，觸發pipeline的fireChannelReadComplete方法。

接下去看看pipeline是如何組織並執行handler對應的方法。

DefaultChannelPipeline

其中DefaultChannelHandlerContext儲存了當前handler的上下文，如channel、pipeline等資訊，預設實現了head和tail。

class DefaultChannelPipeline implements ChannelPipeline {
    final Channel channel; // pipeline所屬的channel
    //head和tail都是handler上下文
    final DefaultChannelHandlerContext head;
    final DefaultChannelHandlerContext tail;
    ...
    public DefaultChannelPipeline(AbstractChannel channel) {
        if (channel == null) {
            throw new NullPointerException("channel");
        }
        this.channel = channel;

        tail = new TailContext(this);
        head = new HeadContext(this);

        head.next = tail;
        tail.prev = head;
    }  
}

12345678910111213141516171819

classDefaultChannelPipelineimplementsChannelPipeline{finalChannel channel;// pipeline所屬的channel//head和tail都是handler上下文finalDefaultChannelHandlerContext head;finalDefaultChannelHandlerContext tail;...publicDefaultChannelPipeline(AbstractChannel

channel){if(channel==null){thrownewNullPointerException("channel");}this.channel=channel;tail=newTailContext(this);head=newHeadContext(this);head.next=tail;tail.prev=head;}}

1、TailContext實現了ChannelOutboundHandler介面。
2、HeadContext實現了ChannelInboundHandler介面。
3、head和tail形成了一個連結串列。

對於Inbound的操作，當channel註冊到selector時，觸發pipeline的fireChannelRegistered，從head開始遍歷，找到實現了ChannelInboundHandler介面的handler，並執行其fireChannelRegistered方法。

@Override
public ChannelPipeline fireChannelRegistered() {
    head.fireChannelRegistered();
    return this;
}

@Override
public ChannelHandlerContext fireChannelRegistered() {
    final DefaultChannelHandlerContext next = findContextInbound();
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeChannelRegistered();
    } else {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                next.invokeChannelRegistered();
            }
        });
    }
    return this;
}

private DefaultChannelHandlerContext findContextInbound() {
    DefaultChannelHandlerContext ctx = this;
    do {
        ctx = ctx.next;
    } while (!(ctx.handler() instanceof ChannelInboundHandler));
    return ctx;
}

private void invokeChannelRegistered() {
    try {
        ((ChannelInboundHandler) handler()).channelRegistered(this);
    } catch (Throwable t) {
        notifyHandlerException(t);
    }
}

1234567891011121314151617181920212223242526272829303132333435363738

@OverridepublicChannelPipeline fireChannelRegistered(){head.fireChannelRegistered();returnthis;}@OverridepublicChannelHandlerContext fireChannelRegistered(){finalDefaultChannelHandlerContext next=findContextInbound();EventExecutor executor=next.executor();if(executor.inEventLoop()){next.invokeChannelRegistered();}else{executor.execute(newRunnable(){@Overridepublicvoidrun(){next.invokeChannelRegistered();}});}returnthis;}privateDefaultChannelHandlerContext findContextInbound(){DefaultChannelHandlerContext ctx=this;do{ctx=ctx.next;}while(!(ctx.handler()instanceofChannelInboundHandler));returnctx;}privatevoidinvokeChannelRegistered(){try{((ChannelInboundHandler)handler()).channelRegistered(this);}catch(Throwablet){notifyHandlerException(t);}}

假如我們通過pipeline的addLast方法新增一個inboundHandler實現。

public class ClientHandler extends ChannelInboundHandlerAdapter {
    @Override  
    public void channelRegistered(ChannelHandlerContext ctx)  
            throws Exception {  
        super.channelRegistered(ctx);  
        System.out.println(" ClientHandler  registered channel ");  
    }  
}

12345678

publicclassClientHandlerextendsChannelInboundHandlerAdapter{@Override publicvoidchannelRegistered(ChannelHandlerContext ctx)throwsException{super.channelRegistered(ctx);System.out.println(" ClientHandler registered channel ");}}

當channel註冊完成時會觸發pipeline的channelRegistered方法，從head開始遍歷，找到ClientHandler，並執行channelRegistered方法。

對於Outbound的操作，則從tail向前遍歷，找到實現ChannelOutboundHandler介面的handler，具體實現和Inbound一樣。

服務啟動過程中，ServerBootstrap在init方法中，會給ServerSocketChannel的pipeline新增ChannelInitializer物件，其中ChannelInitializer繼承ChannelInboundHandlerAdapter，並實現了ChannelInboundHandler介面，所以當ServerSocketChannel註冊到selector之後，會觸發其channelRegistered方法。

public final void channelRegistered(ChannelHandlerContext ctx) throws Exception {
    initChannel((C) ctx.channel());
    ctx.pipeline().remove(this);
    ctx.fireChannelRegistered();
}

public void initChannel(Channel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    ChannelHandler handler = handler();
    if (handler != null) {
        pipeline.addLast(handler);
    }
    pipeline.addLast(new ServerBootstrapAcceptor(
            currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}

123456789101112131415

publicfinalvoidchannelRegistered(ChannelHandlerContext ctx)throwsException{initChannel((C)ctx.channel());ctx.pipeline().remove(this);ctx.fireChannelRegistered();}publicvoidinitChannel(Channel ch)throwsException{ChannelPipeline pipeline=ch.pipeline();ChannelHandler handler=handler();if(handler!=null){pipeline.addLast(handler);}pipeline.addLast(newServerBootstrapAcceptor(currentChildGroup,currentChildHandler,currentChildOptions,currentChildAttrs));}

在initChannel實現中，新增ServerBootstrapAcceptor例項到pipeline中。

ServerBootstrapAcceptor繼承自ChannelInboundHandlerAdapter，負責把接收到的客戶端socketChannel註冊到childGroup中，由childGroup中的eventLoop負責資料處理。

public void channelRead(ChannelHandlerContext ctx, Object msg) {
    final Channel child = (Channel) msg;

    child.pipeline().addLast(childHandler);

    for (Entry, Object> e: childOptions) {
        try {
            if (!child.config().setOption((ChannelOption) e.getKey(), e.getValue())) {
                logger.warn("Unknown channel option: " + e);
            }
        } catch (Throwable t) {
            logger.warn("Failed to set a channel option: " + child, t);
        }
    }

    for (Entry, Object> e: childAttrs) {
        child.attr((AttributeKey) e.getKey()).set(e.getValue());
    }

    try {
        childGroup.register(child).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    forceClose(child, future.cause());
                }
            }
        });
    } catch (Throwable t) {
        forceClose(child, t);
    }
}

1234567891011121314151617181920212223242526272829303132

publicvoidchannelRead(ChannelHandlerContext ctx,Objectmsg){finalChannel child=(Channel)msg;child.pipeline().addLast(childHandler);for(Entry,Object>e:childOptions){try{if(!child.config().setOption((ChannelOption)e.getKey(),e.getValue())){logger.warn("Unknown channel option: "+e);}}catch(Throwablet){logger.warn("Failed to set a channel option: "+child,t);}}for(Entry,Object>e:childAttrs){child.attr((AttributeKey)e.getKey()).set(e.getValue());}try{childGroup.register(child).addListener(newChannelFutureListener(){@OverridepublicvoidoperationComplete(ChannelFuture future)throwsException{if(!future.isSuccess()){forceClose(child,future.cause());}}});}catch(Throwablet){forceClose(child,t);}}

END。

深入淺出Netty：ChannelPipeline

DefaultChannelPipeline

深入淺出Netty：ChannelPipeline

深入淺出Netty：write

深入淺出Netty：NioEventLoop

深入淺出Netty：服務啟動

深入淺出Netty：read

深入淺出Netty：accept

Netty學習：ChannelPipeline

深入淺出Netty記憶體管理：PoolSubpage

深入淺出Netty記憶體管理：PoolChunk

深入淺出Netty記憶體管理：PoolArena

深入淺出Netty記憶體管理：PoolChunkList

深入淺出講解：php的socket通信

深入淺出CSS：Div（一）

下載-深入淺出Netty源碼剖析、Netty實戰高性能分布式RPC、NIO+Netty5各種RPC架構實戰演練三部曲視頻教程

深入淺出MyBatis：JDBC和MyBatis介紹

深入淺出MyBatis：反射和動態代理

深入淺出MyBatis：「映射器」全了解

深入淺出MyBatis：MyBatis與Spring集成及實用場景

【Netty 專欄】深入淺出 Netty write

深入淺出講解：php的socket通訊

深入淺出Netty：ChannelPipeline

DefaultChannelPipeline

相關推薦