前言

本篇文章我們就來(lái)說(shuō)說(shuō)Netty的架構(gòu)設(shè)計(jì)拍顷，解密高并發(fā)之道抚太。學(xué)習(xí)一個(gè)框架之前，我們首先要弄懂它的設(shè)計(jì)原理昔案，然后再進(jìn)行深層次的分析尿贫。

接下來(lái)我們從三個(gè)方面來(lái)分析 Netty 的架構(gòu)設(shè)計(jì)。

Selector 模型

Java NIO 是基于 Selector 模型來(lái)實(shí)現(xiàn)非阻塞的 I/O踏揣。Netty 底層是基于 Java NIO 實(shí)現(xiàn)的庆亡，因此也使用了 Selector 模型。

Selector 模型解決了傳統(tǒng)的阻塞 I/O 編程一個(gè)客戶端一個(gè)線程的問(wèn)題捞稿。Selector 提供了一種機(jī)制又谋，用于監(jiān)視一個(gè)或多個(gè) NIO 通道拼缝，并識(shí)別何時(shí)可以使用一個(gè)或多個(gè) NIO 通道進(jìn)行數(shù)據(jù)傳輸。這樣彰亥，一個(gè)線程就可以管理多個(gè)通道咧七，從而管理多個(gè)網(wǎng)絡(luò)連接。

image-20210804231340262

Selector 提供了選擇執(zhí)行已經(jīng)就緒的任務(wù)的能力任斋。從底層來(lái)看继阻，Selector 會(huì)輪詢 Channel 是否已經(jīng)準(zhǔn)備好執(zhí)行每個(gè) I/O 操作。Selector 允許單線程處理多個(gè) Channel 废酷。Selector 是一種多路復(fù)用的技術(shù)瘟檩。

SelectableChannel

并不是所有的 Channel 都是可以被 Selector 復(fù)用的，只有抽象類 SelectableChannel的子類才能被 Selector 復(fù)用澈蟆。

例如墨辛，FileChannel 就不能被選擇器復(fù)用，因?yàn)?FileChannel 不是SelectableChannel的子類趴俘。

為了與 Selector 一起使用背蟆，SelectableChannel必須首先通過(guò)register方法來(lái)注冊(cè)此類的實(shí)例。此方法返回一個(gè)新的SelectionKey對(duì)象哮幢，該對(duì)象表示Channel已經(jīng)在Selector進(jìn)行了注冊(cè)带膀。向Selector注冊(cè)后，Channel將保持注冊(cè)狀態(tài)橙垢，直到注銷為止垛叨。

一個(gè) Channel 最多可以使用任何一個(gè)特定的 Selector 注冊(cè)一次，但是相同的 Channel 可以注冊(cè)到多個(gè) Selector 上柜某∷栽可以通過(guò)調(diào)用 isRegistered方法來(lái)確定是否向一個(gè)或多個(gè) Selector 注冊(cè)了 Channel。

SelectableChannel可以安全的供多個(gè)并發(fā)線程使用喂击。

Channel 注冊(cè)到 Selector

使用 SelectableChannel的register方法剂癌，可將Channel注冊(cè)到Selector。方法接口源碼如下：

    public final SelectionKey register(Selector sel, int ops)
        throws ClosedChannelException
    {
        return register(sel, ops, null);
    }

    public abstract SelectionKey register(Selector sel, int ops, Object att) throws ClosedChannelException;

其中各選項(xiàng)說(shuō)明如下：

• sel：指定 Channel 要注冊(cè)的 Selector翰绊。
• ops : 指定 Selector需要查詢的通道的操作佩谷。

一個(gè)Channel在Selector注冊(cè)其代表的是一個(gè)SelectionKey事件，SelectionKey的類型包括：

• OP_READ：可讀事件监嗜；值為：1<<0
• OP_WRITE：可寫(xiě)事件谐檀；值為：1<<2
• OP_CONNECT：客戶端連接服務(wù)端的事件(tcp連接)，一般為創(chuàng)建SocketChannel客戶端channel裁奇；值為：1<<3
• OP_ACCEPT：服務(wù)端接收客戶端連接的事件桐猬，一般為創(chuàng)建ServerSocketChannel服務(wù)端channel；值為：1<<4

具體的注冊(cè)代碼如下：

 // 1.創(chuàng)建通道管理器(Selector)
 Selector selector = Selector.open();

 // 2.創(chuàng)建通道ServerSocketChannel
 ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

 // 3.channel要注冊(cè)到Selector上就必須是非阻塞的刽肠，所以FileChannel是不可以使用Selector的溃肪，因?yàn)镕ileChannel是阻塞的
 serverSocketChannel.configureBlocking(false);

 // 4.第二個(gè)參數(shù)指定了我們對(duì) Channel 的什么類型的事件感興趣
 SelectionKey key = serverSocketChannel.register(selector , SelectionKey.OP_READ);

 // 也可以使用或運(yùn)算|來(lái)組合多個(gè)事件免胃，例如
 SelectionKey key = serverSocketChannel.register(selector , SelectionKey.OP_READ | SelectionKey.OP_WRITE);

值得注意的是：一個(gè) Channel 僅僅可以被注冊(cè)到一個(gè)Selector 一次, 如果將 Channel 注冊(cè)到 Selector 多次, 那么其實(shí)就是相當(dāng)于更新 SelectionKey的 interest set。

SelectionKey

Channel 和 Selector 關(guān)系確定后之后惫撰，并且一旦 Channel 處于某種就緒狀態(tài)杜秸，就可以被選擇器查詢到。這個(gè)工作再調(diào)用 Selector 的 select 方法完成润绎。select 方法的作用，就是對(duì)感興趣的通道操作進(jìn)行就緒狀態(tài)的查詢诞挨。

// 當(dāng)注冊(cè)事件到達(dá)時(shí)莉撇，方法返回，否則該方法會(huì)一直阻塞
selector.select();

SelectionKey 包含了 interest 集合惶傻，代表了所選擇的感興趣的事件集合棍郎。可以通過(guò) SelectionKey 讀寫(xiě) interest 集合银室，例如：

// 返回當(dāng)前感興趣的事件列表
int interestSet = key.interestOps();

// 也可通過(guò)interestSet判斷其中包含的事件
boolean isInterestedInAccept  = interestSet & SelectionKey.OP_ACCEPT;
boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT;
boolean isInterestedInRead    = interestSet & SelectionKey.OP_READ;
boolean isInterestedInWrite   = interestSet & SelectionKey.OP_WRITE;    

// 可以通過(guò)interestOps(int ops)方法修改事件列表
key.interestOps(interestSet | SelectionKey.OP_WRITE);

可以看到涂佃，用位與操作 interest 集合和給定的 SelectionKey 常量，可以確定某個(gè)確定的事件是否在 interest 集合中蜈敢。

SelectionKey 包含了ready集合辜荠。ready 集合是通道已經(jīng)準(zhǔn)備就緒的操作的集合。在一次選擇之后抓狭，會(huì)首先訪問(wèn)這個(gè) ready 集合伯病。可以這樣訪問(wèn) ready 集合：

int readySet = key.readyOps();

// 也可通過(guò)四個(gè)方法來(lái)分別判斷不同事件是否就緒
key.isReadable();    //讀事件是否就緒
key.isWritable();    //寫(xiě)事件是否就緒
key.isConnectable(); //客戶端連接事件是否就緒
key.isAcceptable();  //服務(wù)端連接事件是否就緒

我們可以通過(guò)SelectionKey來(lái)獲取當(dāng)前的channel和selector

//返回當(dāng)前事件關(guān)聯(lián)的通道否过，可轉(zhuǎn)換的選項(xiàng)包括:`ServerSocketChannel`和`SocketChannel`
Channel channel = key.channel();

//返回當(dāng)前事件所關(guān)聯(lián)的Selector對(duì)象
Selector selector = key.selector();

可以將一個(gè)對(duì)象或者其他信息附著到 SelectionKey 上午笛，這樣就能方便地識(shí)別某個(gè)特定的通道。

key.attach(theObject);
Object attachedObj = key.attachment();

還可以在用 register()方法向 Selector 注冊(cè) Channel 的時(shí)候附加對(duì)象苗桂。

SelectionKey key = channel.register(selector, SelectionKey.OP_READ, theObject);

遍歷 SelectionKey

一旦調(diào)用了 select 方法药磺，并且返回值表明有一個(gè)或更多個(gè)通道就緒了，然后可以通過(guò)調(diào)用 selector的 selectedKey()方法煤伟，訪問(wèn) SelectionKey 集合中的就緒通道癌佩，如下所示：

Set<SelectionKey> selectionKeys = selector.selectedKeys();

可以遍歷這個(gè)已選擇的鍵集合來(lái)訪問(wèn)就緒的通道，代碼如下：

// 獲取監(jiān)聽(tīng)事件
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeys.iterator();
// 迭代處理
while (iterator.hasNext()) {
    // 獲取事件
    SelectionKey key = iterator.next();
    // 移除事件便锨，避免重復(fù)處理
    iterator.remove();
    // 可連接
    if (key.isAcceptable()) {
        ...
    } 
    // 可讀
    if (key.isReadable()) {
        ...
    }
    //可寫(xiě)
    if(key.isWritable()){
        ...                
    }
}

事件驅(qū)動(dòng)

Netty是一款異步的事件驅(qū)動(dòng)的網(wǎng)絡(luò)應(yīng)用程序框架驼卖。在 Netty 中，事件是指對(duì)某些操作感興趣的事鸿秆。例如酌畜，在某個(gè)Channel注冊(cè)了 OP_READ，說(shuō)明該 Channel 對(duì)讀感興趣卿叽，當(dāng) Channel 中有可讀的數(shù)據(jù)時(shí)桥胞，它會(huì)得到一個(gè)事件的通知恳守。

在 Netty 事件驅(qū)動(dòng)模型中包括以下核心組件。

Channel

Channel（管道）是 Java NIO 的一個(gè)基本抽象贩虾，代表了一個(gè)連接到如硬件設(shè)備催烘、文件、網(wǎng)絡(luò) socket 等實(shí)體的開(kāi)放連接缎罢，或者是一個(gè)能夠完成一種或多種不同的I/O 操作的程序伊群。

回調(diào)

回調(diào) 就是一個(gè)方法，一個(gè)指向已經(jīng)被提供給另外一個(gè)方法的方法的引用策精。這使得后者可以在適當(dāng)?shù)臅r(shí)候調(diào)用前者舰始，Netty 在內(nèi)部使用了回調(diào)來(lái)處理事件；當(dāng)一個(gè)回調(diào)被觸發(fā)時(shí)咽袜，相關(guān)的事件可以被一個(gè)ChannelHandler接口處理丸卷。

例如：在上一篇文章中，Netty 開(kāi)發(fā)的服務(wù)端的管道處理器代碼中询刹，當(dāng)Channel中有可讀的消息時(shí)谜嫉，NettyServerHandler的回調(diào)方法channelRead就會(huì)被調(diào)用。

public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    //讀取數(shù)據(jù)實(shí)際(這里我們可以讀取客戶端發(fā)送的消息)
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("server ctx =" + ctx);
        Channel channel = ctx.channel();
        //將 msg 轉(zhuǎn)成一個(gè) ByteBuf
        //ByteBuf 是 Netty 提供的凹联，不是 NIO 的 ByteBuffer.
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("客戶端發(fā)送消息是:" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("客戶端地址:" + channel.remoteAddress());
    }

    //處理異常, 一般是需要關(guān)閉通道
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

Future

Future 可以看作是一個(gè)異步操作的結(jié)果的占位符沐兰；它將在未來(lái)的某個(gè)時(shí)刻完成，并提供對(duì)其結(jié)果的訪問(wèn)蔽挠，Netty 提供了 ChannelFuture 用于在異步操作的時(shí)候使用僧鲁，每個(gè) Netty 的出站 I/O 操作都將返回一個(gè) ChannelFuture(完全是異步和事件驅(qū)動(dòng)的)。

以下是一個(gè) ChannelFutureListener使用的示例象泵。

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ChannelFuture future = ctx.channel().close();
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                //..
            }
        });
    }

事件及處理器

在 Netty 中事件按照出/入站數(shù)據(jù)流進(jìn)行分類：

入站數(shù)據(jù)或相關(guān)狀態(tài)更改觸發(fā)的事件包括：

• 連接已被激活或者失活寞秃。
• 數(shù)據(jù)讀取。
• 用戶事件偶惠。
• 錯(cuò)誤事件春寿，

出站事件是未來(lái)將會(huì)出發(fā)的某個(gè)動(dòng)作的操作結(jié)果：

• 打開(kāi)或者關(guān)閉到遠(yuǎn)程節(jié)點(diǎn)的連接。
• 將數(shù)據(jù)寫(xiě)或者沖刷到套接字忽孽。

每個(gè)事件都可以被分發(fā)給ChannelHandler類中的某個(gè)用戶實(shí)現(xiàn)的方法绑改。如下圖展示了一個(gè)事件是如何被一個(gè)這樣的ChannelHandler鏈所處理的。

image-20210805153230027

ChannelHandler 為處理器提供了基本的抽象兄一，可理解為一種為了響應(yīng)特定事件而被執(zhí)行的回調(diào)厘线。

責(zé)任鏈模式

責(zé)任鏈模式(Chain of Responsibility Pattern)是一種行為型設(shè)計(jì)模式，它為請(qǐng)求創(chuàng)建了一個(gè)處理對(duì)象的鏈出革。其鏈中每一個(gè)節(jié)點(diǎn)都看作是一個(gè)對(duì)象造壮，每個(gè)節(jié)點(diǎn)處理的請(qǐng)求均不同，且內(nèi)部自動(dòng)維護(hù)一個(gè)下一節(jié)點(diǎn)對(duì)象。當(dāng)一個(gè)請(qǐng)求從鏈?zhǔn)降氖锥税l(fā)出時(shí)耳璧，會(huì)沿著鏈的路徑依次傳遞給每一個(gè)節(jié)點(diǎn)對(duì)象成箫，直至有對(duì)象處理這個(gè)請(qǐng)求為止。

責(zé)任鏈模式的重點(diǎn)在這個(gè) "鏈"上旨枯，由一條鏈去處理相似的請(qǐng)求蹬昌，在鏈中決定誰(shuí)來(lái)處理這個(gè)請(qǐng)求，并返回相應(yīng)的結(jié)果攀隔。在Netty中皂贩，定義了ChannelPipeline接口用于對(duì)責(zé)任鏈的抽象。

責(zé)任鏈模式會(huì)定義一個(gè)抽象處理器（Handler）角色昆汹，該角色對(duì)請(qǐng)求進(jìn)行抽象明刷，并定義一個(gè)方法來(lái)設(shè)定和返回對(duì)下一個(gè)處理器的引用。在Netty中筹煮，定義了ChannelHandler接口承擔(dān)該角色。

責(zé)任鏈模式的優(yōu)缺點(diǎn)

優(yōu)點(diǎn)：

• 發(fā)送者不需要知道自己發(fā)送的這個(gè)請(qǐng)求到底會(huì)被哪個(gè)對(duì)象處理掉居夹，實(shí)現(xiàn)了發(fā)送者和接受者的解耦败潦。
• 簡(jiǎn)化了發(fā)送者對(duì)象的設(shè)計(jì)。
• 可以動(dòng)態(tài)的添加節(jié)點(diǎn)和刪除節(jié)點(diǎn)准脂。

缺點(diǎn)：

• 所有的請(qǐng)求都從鏈的頭部開(kāi)始遍歷劫扒，對(duì)性能有損耗。
• 不方便調(diào)試狸膏。由于該模式采用了類似遞歸的方式沟饥，調(diào)試的時(shí)候邏輯比較復(fù)雜。

使用場(chǎng)景：

• 一個(gè)請(qǐng)求需要一系列的處理工作湾戳。
• 業(yè)務(wù)流的處理贤旷，例如文件審批。
• 對(duì)系統(tǒng)進(jìn)行擴(kuò)展補(bǔ)充砾脑。

ChannelPipeline

Netty 的ChannelPipeline設(shè)計(jì)幼驶，就采用了責(zé)任鏈設(shè)計(jì)模式， 底層采用雙向鏈表的數(shù)據(jù)結(jié)構(gòu),韧衣，將鏈上的各個(gè)處理器串聯(lián)起來(lái)盅藻。

客戶端每一個(gè)請(qǐng)求的到來(lái)，Netty都認(rèn)為漓帚，ChannelPipeline中的所有的處理器都有機(jī)會(huì)處理它屏积，因此拒逮，對(duì)于入棧的請(qǐng)求，全部從頭節(jié)點(diǎn)開(kāi)始往后傳播假残，一直傳播到尾節(jié)點(diǎn)(來(lái)到尾節(jié)點(diǎn)的msg會(huì)被釋放掉)。

image

入站事件：通常指 IO 線程生成了入站數(shù)據(jù)（通俗理解：從 socket 底層自己往上冒上來(lái)的事件都是入站）炉擅。
比如EventLoop收到selector的OP_READ事件守问，入站處理器調(diào)用socketChannel.read(ByteBuffer)接受到數(shù)據(jù)后匀归，這將導(dǎo)致通道的ChannelPipeline中包含的下一個(gè)中的channelRead方法被調(diào)用。

出站事件：通常指 IO 線程執(zhí)行實(shí)際的輸出操作（通俗理解：想主動(dòng)往 socket 底層操作的事件的都是出站）耗帕。
比如bind方法用意時(shí)請(qǐng)求server socket綁定到給定的SocketAddress穆端，這將導(dǎo)致通道的ChannelPipeline中包含的下一個(gè)出站處理器中的bind方法被調(diào)用。

將事件傳遞給下一個(gè)處理器

處理器必須調(diào)用ChannelHandlerContext中的事件傳播方法仿便，將事件傳遞給下一個(gè)處理器体啰。

入站事件和出站事件的傳播方法如下圖所示：

image

以下示例說(shuō)明了事件傳播通常是如何完成的：

public class MyInboundHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Connected!");
        ctx.fireChannelActive();
    }
}

public class MyOutboundHandler extends ChannelOutboundHandlerAdapter {

    @Override
    public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        System.out.println("Closing...");
        ctx.close(promise);
    }
}

總結(jié)

正是由于 Netty 的分層架構(gòu)設(shè)計(jì)非常合理，基于 Netty 的各種應(yīng)用服務(wù)器和協(xié)議棧開(kāi)發(fā)才能夠如雨后春筍般得到快速發(fā)展嗽仪。

結(jié)尾

我是一個(gè)正在被打擊還在努力前進(jìn)的碼農(nóng)荒勇。如果文章對(duì)你有幫助，記得點(diǎn)贊闻坚、關(guān)注喲沽翔，謝謝！

標(biāo)簽: [Netty]
[圖片上傳失敗...(image-43246d-1628754037255)]