Netty源碼分析之ChannelPipeline—入站事件的傳播

以前的文章中咱們說過ChannelPipeline做爲Netty中的數據管道，負責傳遞Channel中消息的事件傳播，事件的傳播分爲入站和出站兩個方向，分別通知ChannelInboundHandler與ChannelOutboundHandler來觸發對應事件。這篇文章咱們先對Netty中入站事件的傳播，也就是ChannelInboundHandler進行下分析：

一、入站事件傳播示例

咱們經過一個簡單的例子看下ChannelPipeline中入站事件channelRead的傳播

public class ServerApp { public static void main(String[] args) { EventLoopGroup boss = new NioEventLoopGroup(); EventLoopGroup work = new NioEventLoopGroup(2); try { ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.group(boss, work).channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<SocketChannel>() { @Override public void initChannel(SocketChannel ch) throws Exception { ChannelPipeline p = ch.pipeline(); // p.addLast(new LoggingHandler(LogLevel.INFO)); // 向ChannelPipeline中添加自定義channelHandler
                            p.addLast(new ServerHandlerA()); p.addLast(new ServerHandlerB()); p.addLast(new ServerHandlerC()); } }); bootstrap.bind(8050).sync(); } catch (Exception e) { // TODO: handle exception
 } } } public class ServerHandlerA  extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object object) { System.out.println(this.getClass().getName() + "--"+object.toString()); ctx.fireChannelRead(object); } @Override public void channelActive(ChannelHandlerContext ctx) { ctx.channel().pipeline().fireChannelRead("hello word"); } } public class ServerHandlerB extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object object) { System.out.println(this.getClass().getName() + "--"+object.toString()); ctx.fireChannelRead(object); } } public class ServerHandlerC extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object object) { System.out.println(this.getClass().getName() + "--"+object.toString()); ctx.fireChannelRead(object); } }

客戶端鏈接服務後可看到輸出結果

io.netty.example.echo.my.ServerHandlerA--hello word io.netty.example.echo.my.ServerHandlerB--hello word io.netty.example.echo.my.ServerHandlerC--hello word

經過輸出結果咱們能夠看到，消息會根據向ChannelPipeline中添加自定義channelHandler的順序傳遞，並經過實現channelRead接口處理消息接收事件的。在例子中channelRead事件的傳遞是經過ctx.fireChannelRead(object)方法實現，接下來咱們就從這裏入手看下ChannelPipeline事件傳遞的具體實現。

二、channelRead事件的傳播

首先這裏須要注意的是咱們例子中第一個節點的傳遞與實際應用中入站數據的傳遞是經過ChannelPipeline的fireChannelRead方法實現的，由於在實際的應用中，入站事件的傳遞是由NioUnsafe的read接口實現發起的，須要保證消息是從head結點開始傳遞的，例子中是爲了模擬這一過程。

ctx.channel().pipeline().fireChannelRead("hello word");

 @Override public final ChannelPipeline fireChannelRead(Object msg) { AbstractChannelHandlerContext.invokeChannelRead(head, msg);//默認傳入head節點
        return this; } 

進入invokeChannelRead方法內部看下具體實現；

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) { //ObjectUtil.checkNotNull 判斷傳入的消息數據是否爲空 //next.pipeline.touch 對消息類型進行判斷
        final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next); EventExecutor executor = next.executor();//獲取ChannelHandlerContext對應的線程
        if (executor.inEventLoop()) {//是否爲當前線程
            next.invokeChannelRead(m);//調用ChannelHandlerContext中invokeChannelRead的回調方法
        } else { executor.execute(new Runnable() {//若是線程不是當前線程
 @Override public void run() { next.invokeChannelRead(m); } }); } }

其中invokeChannelRead方法會獲取該ChannelHandlerContext所封裝的handler實現；

private void invokeChannelRead(Object msg) { if (invokeHandler()) { try { //獲取封裝的ChannelInboundHandler實現，並調用咱們實現的channelRead方法，
                ((ChannelInboundHandler) handler()).channelRead(this, msg); } catch (Throwable t) { notifyHandlerException(t); } } else { fireChannelRead(msg); } }

前面咱們知道首先傳入的ChannelPipeline中ChannelHandlerContext鏈表的head頭部節點HeadContext，看下其channelRead的方法實現；

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

調用當前ChannelHandlerContext的fireChannelRead方法，進入ctx.fireChannelRead(object)方法內部看下具體的源碼實現；

 @Override public ChannelHandlerContext fireChannelRead(final Object msg) { //開始消息傳遞，findContextInbound方法按順序獲取當前ChannelHandlerContext的next節點
 invokeChannelRead(findContextInbound(), msg); return this; }

findContextInbound方法獲取的是HeadContext的下一個節點，也就是咱們例子中向ChannelPipeline中添加自定義ServerHandlerA；

到這裏其實就能夠看出Pipeline中channelRead事件的傳播主要就是經過ctx.fireChannelRead(msg)，獲取當前ChannelHandlerContext下一個節點中封裝的ChannelInboundHandler來實現的，最後一步一步傳遞到Tail尾部節點。

三、資源的釋放及SimpleChannelInboundHandler

Netty中對象的生命週期由它們的引用計數管理的，爲保證入站對象資源被釋放，咱們須要經過ReferenceCountUtil.release方法減小引用計數，確保對象的的最終計數器最後被置爲0，從而被回收釋放。咱們看下Netty在入站事件中默認是如何減小引用計數的。

第一種方法，若是咱們跟上面示例同樣，在實現的每個ChannelInboundHandler中都調用了ctx.fireChannelRead(msg)，最後消息會被傳遞到Tail尾節點，咱們看下Tail節點中的channelRead方法

 @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { 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); } }

Tail節點的channelRead方法最終會調用ReferenceCountUtil.release方法來減小引用計數的，因此若是你在處理入站消息的過程當中沒有增長引用而且經過ctx.fireChannelRead(msg)方法把消息傳到了Tail節點，你就不須要本身顯式調用ReferenceCountUtil.release方法了。

其次若是繼承的是SimpleChannelInboundHandler，能夠看到SimpleChannelInboundHandler的channelRead方法實現中也已經調用了ReferenceCountUtil.release方法來減小引用計數；

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

因此關於入站消息的資源釋放方式總結以下：

• 一、繼承ChannelInboundHandlerAdapter ，在channelRead的方法實現中調用ctx.fireChannelRead(object)方法，把消息一直向下傳遞，直到傳遞到Tail尾部節點，由Tail節點執行 ReferenceCountUtil.release來減小計數器，保證資源釋放；
• 二、繼承SimpleChannelInboundHandler，SimpleChannelInboundHandler自己的ChannelRead方法中會執行 ReferenceCountUtil.release來減小引用；
• 三、若是以上兩點都沒有作到，那就須要手動調用ReferenceCountUtil.release來減小引用來釋放資源；

 

到這裏咱們基本瞭解了ChannelPipeline中入站事件是如何傳播與相應的的，以及Netty中入站消息的資源釋放機制。其中若有不足與不正確的地方還望指出與海涵。

 

關注微信公衆號，查看更多技術文章。

 

 

原文出處：https://www.cnblogs.com/dafanjoy/p/12274701.html