Netty ChannelPipeline 數據管道

1. 前言

前面，我們也提到了 ChannelPipeline，它是管道或者說管理 Handler 的集合，很多同學很容易搞混 Channel、ChannelPipeline 和 ChannelHandler 之間關系。

本節(jié)內容我們需要理清并且掌握以下知識點：

1. Channel、ChannelPipeline、ChannelHandler 之間的關系；
2. 了解 ChannelPipeline 如何管理 ChannelHandler。

2. 三者之間的關系

Channel 是一個連接通道，客戶端和服務端連接成功之后，會維持一個 Channel，可以通過 Channel 來發(fā)送數據。Channel 有且僅有一個 ChannelPipeline 與之相對應，ChannelPipeline 又維護著一個由多個 ChannelHandlerContext 組成的雙向鏈表，ChannelHandlerContext 又關聯著一個 ChannelHandler。

它們之間的關系，大概如下圖所示：

3. ChannelPipeline 核心方法

ChannelPipeline 的最常用方法：

實例：

ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap
    .group(bossGroup, workerGroup)
    .channel(NioServerSocketChannel.class)
    .childHandler(new ChannelInitializer<NioSocketChannel>() {
        protected void initChannel(NioSocketChannel ch) {
            ch.pipeline().addLast(new Handler1());
            ch.pipeline().addLast(new Handler2());
            ch.pipeline().addLast(new Handler3());
            ch.pipeline().addLast(new Handler4());
        }
    });

總結，ChannelPipeline 的用法比較固定，雖然方法很多，但是一般常用的就是 addLast。

4. ChannelPipeline 管理鏈表

實例：

//1.創(chuàng)建ChannelPipeline
ChannelPipeline pipeline = ch.pipeline();

//2.創(chuàng)建Handler
FirstHandler firstHandler = new FirstHandler();
SecondHandler secondHandler=new SecondHandler();
ThirdHandler thirdHandler=new ThirdHandler();
FourthHandler fourthHandler=new FourthHandler();

//3.操作
pipeline.addLast("handler1", firstHandler);
pipeline.addFirst("handler2", secondHandler);//在最開始添加
pipeline.addLast("handler3", thirdHandler);//在最后添加
pipeline.remove("handler3"); //根據名稱刪除
pipeline.remove(firstHandler);//根據對象刪除
pipeline.replace("handler2", "handler4", fourthHandler);//替換

輸出結果：

FourthHandler

5. 入站和出站執(zhí)行順序

在真實的項目開發(fā)當中，inboundHandler 和 outboundHandler 都是多個的，一般是一個業(yè)務處理對應一個 Handler。那么多個的情況下，Pipeline 的執(zhí)行順序又是怎么樣的呢？

5.1 Inbound 不往下傳遞

實例：

ch.pipeline().addLast(new InboundHandler1());
ch.pipeline().addLast(new InboundHandler2());

public class InboundHandler1 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound1>>>>>>>>>");
    }
}

public class InboundHandler2 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound2>>>>>>>>>");
    }
}

執(zhí)行結果：

inbound1>>>>>>>>>

思考：為什么不執(zhí)行 InboundHandler2 呢？

原因：InboundHandler1 沒有手工往下傳遞執(zhí)行。

5.2 Inbound 流轉順序

實例：

ch.pipeline().addLast(new InboundHandler1());
ch.pipeline().addLast(new InboundHandler2());

public class InboundHandler1 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound1>>>>>>>>>");
        
        //往下傳遞
        super.channelRead(ctx, msg);
    }
}

public class InboundHandler2 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound2>>>>>>>>>");
    }
}

執(zhí)行結果：

inbound1>>>>>>>>>
inbound2>>>>>>>>>

InboundHandler 之間可以通過 super.channelRead(ctx, msg); 往下傳遞。

5.3 Inbound 執(zhí)行順序

實例：

ch.pipeline().addLast(new InboundHandler1());
ch.pipeline().addLast(new InboundHandler2());

public class InboundHandler1 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //1.往下傳遞
        super.channelRead(ctx, msg);
        //2.打印信息
        System.out.println("inbound1>>>>>>>>>");
    }
}

public class InboundHandler2 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound2>>>>>>>>>");
    }
}

執(zhí)行結果：

inbound2>>>>>>>>>
inbound1>>>>>>>>>

InboundHandler1 先往下傳遞，在執(zhí)行自己的業(yè)務，那么 InboundHandler2 就會比 InboundHandler1 先執(zhí)行。

總結：Inbound 是按順序進行傳遞，但是邏輯的執(zhí)行并非是按順序執(zhí)行，而是由 super.channelRead(ctx, msg); 去決定。

5.4 流轉到 Outbound

InboundHandler 往 OutboundHandler 流轉，需要手工調用 ctx.channel().writeAndFlush()，否則無法執(zhí)行 OutboundHandler 的業(yè)務邏輯。

實例：

public class InboundHandler2 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound2>>>>>>>>>");
		//傳遞到OutboundHandler
        ctx.channel().writeAndFlush("hello world");
    }
}

5.5 Outbound 內部流轉

跟 InboundHandler 一樣，需要手工往下傳遞，否則無法流轉到下一個 OutboundHandler。

實例：

public class OutboundHandler2 extends ChannelOutboundHandlerAdapter {
    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        System.out.println("outbound2>>>>>>>>>");
		//往下流轉
        super.write(ctx, msg, promise);
    }
}

總結：OutboundHandler 是按逆向來流轉，但是業(yè)務邏輯的執(zhí)行順序則是由 super.write(ctx, msg, promise); 決定。

5.6 ctx.writeAndFlush 和 ctx.channel ().writeAndFlush 的區(qū)別

很多同學很容易遇到以下問題，并且會想不通。

實例：

ch.pipeline().addLast(new InboundHandler1());
ch.pipeline().addLast(new InboundHandler2());
ch.pipeline().addLast(new OutboundHandler1());
ch.pipeline().addLast(new OutboundHandler2());

InboundHandler2 流轉代碼

public class InboundHandler2 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound2>>>>>>>>>");
		//流轉到OutboundHandler2
        ctx.channel().writeAndFlush("hello world");
    }
}

執(zhí)行結果：

inbound1>>>>>>>>>
inbound2>>>>>>>>>
outbound2>>>>>>>>>
outbound1>>>>>>>>>

修改 InboundHandler2 流轉代碼

public class InboundHandler2 extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("inbound2>>>>>>>>>");
		//【注意，調整這里了】
        ctx.writeAndFlush("hello world");
    }
}

執(zhí)行結果

inbound1>>>>>>>>>
inbound2>>>>>>>>>

思考：為什么這里使用 ctx.writeAndFlush 就流程不下去了呢？

ctx.writeAndFlush(); 最終源碼

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

通過源碼，我們發(fā)現它是從當前 InboundHandler 開始往前執(zhí)行。

ctx.channel().writeAndFlush(); 最終源碼

public final ChannelFuture writeAndFlush(Object msg) {
    return this.tail.writeAndFlush(msg);
}

通過源碼，我們發(fā)現它是從鏈表的最后一個節(jié)點開始往前面執(zhí)行。

總結，如果是 OutboundHandler 放在 InboundHandler 之后，使用不同的 writeAndFlush 則得到的結果不一樣。

5.7 規(guī)律總結

Inbound 的順序

1. 流轉順序： 多個 Inbound 不會自動往下流轉，需要手工調用 ctx.fireChannelRead(msg); 才能流轉到下一個；
2. 執(zhí)行順序： 業(yè)務邏輯的執(zhí)行順序，則根據 ctx.fireChannelRead(msg); 和邏輯的先后順序所決定；
3. Inbound 往 Outbound 流轉，則需要手工 ctx.channel().writeAndFlush()。

Outbound 的順序

1. 流轉順序： 多個 Outbound 不會自動往下流轉，需要手工調用 ctx.write(msg, promise); 才能流轉到下一個；
2. 執(zhí)行順序： 業(yè)務邏輯的執(zhí)行順序，則根據 ctx.write(msg, promise); 和邏輯的先后順序所決定。

6. 小結

本文主要講解的知識點

1. Channel、ChannelPipeline、ChannelHandlerContext、ChannelHandler 之間的關系；
2. ChannelPipeline 的核心方法，以及它是如何管理 Handler 的，主要通過 addLast () 去組裝 Handler；
3. 入站和出站的流轉順序和業(yè)務邏輯的執(zhí)行順序。