前言

之前總結過NIO的三個組件，其實對于Netty來說，如果要理解后續的編解碼器已經實際的Netty應用場景，還有幾個Netty的組件需要了解。這篇博客依舊參考《Netty、Redis、Zookeeper高并發實戰》一書進行總結。

channel

Netty中，channel是核心概念，表示一個網絡連接通道，類似于網絡的socket，Netty關于channel有一個頂層沖虛向類——AbstractChannel，其中有一個構造函數如下：

/**
 * Creates a new instance.
 *
 * @param parent
 *        the parent of this channel. {@code null} if there's no parent.
 */
protected AbstractChannel(Channel parent) {
    this.parent = parent;//父通道
    id = newId();
    unsafe = newUnsafe();
    pipeline = newChannelPipeline();//每一個通道，其實都維護一個類似于流水線的東西，這個后面會總結
}

這里有兩個屬性需要說明一下，一個是parent，另一個是pipeline

parent表示通道的父通道，對于連接建立的通道（比如NioServerScoketChannel）來說，其父通道為null。而對于每一條傳輸數據的通道（比如NioSocketChannel），其父通道就是建立連接的通道。

如果簡單一點理解，channel就是連接網絡的通道。

EmbeddedChannel

這里說一下EmbeddedChannel，我們在開發Netty的時候，大部分時候都是在開發Inbound入站處理器，開發完成之后，如果需要測試相關邏輯，我們需要將處理器加入到通道流水線中，然后啟動服務端和客戶端。每次修改一個業務，都要啟動服務端和客戶端，這個過程非常繁瑣并且浪費時間。于是Netty就給我們提供了一個專用的通道——EmbeddedChannel。

Handler

在經典的反應堆模式中，我們知道每一個注冊在Selector上的網絡事件都會被分發給指定的handler去處理，在Netty中也是一樣，Netty中的Handler就是處理不同的網絡事件。針對入站和出站，有著不同的處理器。

入站處理——數據流動方向為Netty的內部channel到ChannelInboundHandler。

出站處理——數據流動方向為ChannelOutboundHandler到Netty的channel。

但是實際過程中，Netty給我們提供了很多默認的handler實現，一般不需要我們直接去繼承ChannelHandler，畢竟ChannelHandler是一個接口，如果我們每次新增一個Handler，都需要實現ChannelHandler中的每個方法，這樣很麻煩，于是Netty就提供了一個適配器，提供了ChannelHandler的默認實現，更多的時候，我們可以直接繼承ChannelHandlerAdapter即可。具體的類結構如下所示：

方法名稱	作用
channelRegistered	通道注冊完成之后，Netty底層會調用fireChannelRegistered，觸發通道注冊事件，通道就會啟動該入站操作的流水線處理。在通道注冊過的Handler處理器中的channelRegistered方法就會被調用。
channelActive	通道**的時候，在通道注冊過的Handler中的channelActive方法就會被調用
channelRead	通道可讀的時候，在通道注冊過的Handler中的channelRead方法就會被調用
channelReadComplete	通道讀取結束的時候，在通道注冊過的Handler中的channelReadComplete就會被調用
channelInactive	通道連接不可用的時候，在通道注冊的Handler中的channelInactive就會被調用
exceptionCaught	通道處理過程發生異常的時候，在通道注冊的Handler中的exceptionCaught就會被調用。

ChannelInitializer通道初始化處理器

其實每一條channel都有一條Handler業務處理流水線，這就是我們后續要總結的pipeline ，但是負責裝配自己的Handler處理器這個工作是發生在通道開始工作之前。通道的初始化類，就可以向流水線中裝配業務處理類。

serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        socketChannel.pipeline().addLast(new FixedLengthFrameDecoder(requestLength));
        socketChannel.pipeline().addLast(new FixLengthTimeServerHandler());
    }
});

上述的代碼中就在SocketChannel初始化的時候，增加了兩個處理器。

pipeline

Netty的業務處理器流水線ChannelPipeline是基于責任鏈設計模式來設計的。針對每一個channel，都會有一個業務流水線與其綁定，這個流水線中包含了所有的入站和出站處理器，這個業務流水線就是pipeline，pipeline，channel，ChannelHandler三者的關系如下所示。

每個處理器被加入到pipeline的時候，都會有一個ChannelHandlerContext與之綁定。

總體來說，Channel、Handler、pipeline、ChannelHandlerContext幾者的關系為，Channel通道擁有一個ChannelPipeline流水線，每個流水線節點為一個ChannelHandlerContext通道處理器上下對象，每一個上下文中包含了一個ChannelHandler處理器。在ChannelHandler通道處理器的入站/出站處理方法中，Netty會傳遞ChannelHandlerContext，通過ChannelHandlerContext可以獲取pipeline的實例或者Channel的實例。

流水線的傳遞與截斷

Inbound的事件處理順序

在每次初始化增加channel處理器的時候，這些處理器會自動成為pipeline中的節點，這里直接上實例代碼

/**
 * autor:liman
 * createtime:2020/9/19
 * comment:simpleInHandlerA
 */
@Slf4j
public class SimpleInHandlerA extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("this is A simpleInHandler read method，message:{}",msg);
        super.channelRead(ctx, msg);
    }
}

/**
 * autor:liman
 * createtime:2020/9/19
 * comment:simpleInHandlerB
 */
@Slf4j
public class SimpleInHandlerB extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("this is B simpleInHandler read method message:{}",msg);
        super.channelRead(ctx, msg);
    }
}

/**
 * autor:liman
 * createtime:2020/9/19
 * comment:simpleInHandlerC
 */
@Slf4j
public class SimpleInHandlerC extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("this is C simpleInHandler read method message:{}",msg);
        super.channelRead(ctx, msg);
    }
}

使用EmbeddedChannel進行測試

/**
 * autor:liman
 * createtime:2020/9/19
 * comment: InboundHandler的處理器測試
 */
@Slf4j
public class TestPipelineInbound {

    public static void main(String[] args) {
        ChannelInitializer<EmbeddedChannel> channelInitializer = new ChannelInitializer<EmbeddedChannel>() {
            @Override
            protected void initChannel(EmbeddedChannel ch) throws Exception {
                ch.pipeline().addLast(new SimpleInHandlerA());
                ch.pipeline().addLast(new SimpleInHandlerB());
                ch.pipeline().addLast(new SimpleInHandlerC());
            }
        };

        EmbeddedChannel embeddedChannel = new EmbeddedChannel(channelInitializer);
        ByteBuf buf = Unpooled.buffer();
        buf.writeInt(10);
        embeddedChannel.writeInbound(10);//向通道中寫一個報文
    }
}

運行結果：

可以看到調用順序為A->B->C

Outbound的事件處理順序

3個OutboundHandler的業務代碼

/**
 * autor:liman
 * createtime:2020/9/19
 * comment:SimpleOutHandlerA
 */
@Slf4j
public class SimpleOutHandlerA extends ChannelOutboundHandlerAdapter {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        log.info("出站處理器A，this is simple out handler A，write method：{}",msg);
        super.write(ctx, msg, promise);
    }
}

/**
 * autor:liman
 * createtime:2020/9/19
 * comment:
 */
@Slf4j
public class SimpleOutHandlerB extends ChannelOutboundHandlerAdapter {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        log.info("出站處理器B，this is simple out handler B，write method：{}",msg);
        super.write(ctx, msg, promise);
    }
}

/**
 * autor:liman
 * createtime:2020/9/19
 * comment:
 */
@Slf4j
public class SimpleOutHandlerC extends ChannelOutboundHandlerAdapter {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        log.info("出站處理器C，this is simple out handler C，write method：{}",msg);
        super.write(ctx, msg, promise);
    }
}

測試代碼

/**
 * autor:liman
 * createtime:2020/9/19
 * comment:
 */
@Slf4j
public class TestPipelineOutBound {

    public static void main(String[] args) {
        ChannelInitializer<EmbeddedChannel> channelInitializer = new ChannelInitializer<EmbeddedChannel>() {
            @Override
            protected void initChannel(EmbeddedChannel ch) throws Exception {
                ch.pipeline().addLast(new SimpleOutHandlerA());
                ch.pipeline().addLast(new SimpleOutHandlerB());
                ch.pipeline().addLast(new SimpleOutHandlerC());
            }
        };

        EmbeddedChannel embeddedChannel = new EmbeddedChannel(channelInitializer);
        ByteBuf buf = Unpooled.buffer();
        buf.writeInt(10);
        embeddedChannel.writeOutbound(10);//向通道中寫一個報文
    }
}

運行結果

可以看到調用順序是C->B->A。

如果需要截斷流水線的處理，只需要在Handler中去掉調用super的read/write方法。具體實例可以直接注釋相關代碼然后測試即可。

總結

簡單總結了Channel，ChannelHandler，ChannelHandlerContext，Pipeline的關系。同時需要注意的是channelHandler出站和入站處理器的調用方向是不同的。