Netty實(shí)戰(zhàn) IM即時(shí)通訊系統(tǒng)（十一）pipeline與channelHandler

零、 目錄

IM系統(tǒng)簡(jiǎn)介

• Netty 簡(jiǎn)介
• Netty 環(huán)境配置
• 服務(wù)端啟動(dòng)流程
• 客戶(hù)端啟動(dòng)流程
• 實(shí)戰(zhàn)： 客戶(hù)端和服務(wù)端雙向通信
• 數(shù)據(jù)傳輸載體ByteBuf介紹
• 客戶(hù)端與服務(wù)端通信協(xié)議編解碼
• 實(shí)現(xiàn)客戶(hù)端登錄
• 實(shí)現(xiàn)客戶(hù)端與服務(wù)端收發(fā)消息
• pipeline與channelHandler
• 構(gòu)建客戶(hù)端與服務(wù)端pipeline
• 拆包粘包理論與解決方案
• channelHandler的生命周期
• 使用channelHandler的熱插拔實(shí)現(xiàn)客戶(hù)端身份校驗(yàn)
• 客戶(hù)端互聊原理與實(shí)現(xiàn)
• 群聊的發(fā)起與通知
• 群聊的成員管理（加入與退出，獲取成員列表）
• 群聊消息的收發(fā)及Netty性能優(yōu)化
• 心跳與空閑檢測(cè)
• 總結(jié)
• 擴(kuò)展

一、 簡(jiǎn)介

這一小節(jié)中 ， 我們來(lái)學(xué)習(xí)Netty 中一大核心組件： pipeline 與 channelHandler

上一小節(jié)最后 ， 我們提出： 如何避免switch-case 泛濫？ ， 我們注意到， 不管是服務(wù)端還是客戶(hù)端 ， 處理流程大致分為以下步驟

我們把這三類(lèi)邏輯都寫(xiě)在一個(gè)類(lèi)里面， 客戶(hù)端寫(xiě)在 ClientHandler , 服務(wù)端寫(xiě)在 ServerHandler , 如果要做功能的擴(kuò)展 （比如 ， 我們要校驗(yàn)?zāi)?shù) ， 或者其他特殊邏輯） ， 只能在一個(gè)類(lèi)里面去修改 ， 這個(gè)類(lèi)就會(huì)變得越來(lái)越臃腫。

另外 ， 我們注意到， 每次發(fā)指令數(shù)據(jù)包都要是都調(diào)用編碼器編碼成byteBuf ， 對(duì)于這類(lèi)場(chǎng)景的編碼優(yōu)化， 我們能想到的辦法自然是模塊化處理 ， 不同的邏輯放置到單獨(dú)的類(lèi)中來(lái)處理 ， 最后將這些邏輯串聯(lián)起來(lái) ， 形成一個(gè)完整的邏輯處理鏈 。

Netty中的pipeline 與channelHandler 正是來(lái)解決這個(gè)問(wèn)題的： 他通過(guò)責(zé)任鏈設(shè)計(jì)模式來(lái)組織代碼邏輯 ， 并且能夠支持邏輯的動(dòng)態(tài)添加和刪除 ， Netty能夠支持各類(lèi)協(xié)議的支持和擴(kuò)展 ， 比如： HTTP , WebSocket ，Redis 靠的就是pipeline與channelHanler。

二、 pipeline 與channelHandler 的構(gòu)成

無(wú)論是從服務(wù)端來(lái)看 ， 還是從客戶(hù)端來(lái)看 ， 在Netty整個(gè)框架中， 一條連接對(duì)應(yīng)著一個(gè)channel ,這個(gè)channel 所有的處理邏輯都在一個(gè)叫做ChannelPipeline的對(duì)象里面 ， ChannelPipeline 是一個(gè)雙向鏈表結(jié)構(gòu) ， 他和Channel之間是一對(duì)一的關(guān)系

ChannelPipeline 里面每個(gè)節(jié)點(diǎn)都是一個(gè)ChannelHandlerContext 對(duì)象 ， 這個(gè)對(duì)象能夠拿到和Channel相關(guān)的上下文信息 ， 然后這個(gè)對(duì)象抱著一個(gè)重要的對(duì)象 ， 那就是邏輯處理器 ChannelHandler

接下啦我們來(lái)看一下ChannelHandler有哪些分類(lèi)

三、 ChannelHandler 的分類(lèi)

可以考到ChannelHandler 有兩大子接口

第一個(gè)子接口是ChannelInboundHandler , 從字面意思也可以猜到，他是處理讀數(shù)據(jù)的邏輯 ， 比如： 我們?cè)谝欢巫x到一段數(shù)據(jù) ， 首先要解析這段數(shù)據(jù) ， 然后對(duì)這些數(shù)據(jù)做一些邏輯處理 ， 最終把響應(yīng)寫(xiě)到對(duì)端 ， 在開(kāi)始組裝響應(yīng)之前的邏輯 ， 都可以放在ChannelInboundHandler 中處理 ， 他的一個(gè)最重要的方法就是 channelRead , 讀者可以將ChannelInboundHandler中的邏輯處理過(guò)程與TCP的七層協(xié)議的解析連接起來(lái) ， 收到的數(shù)據(jù)一層一層從物理層傳送到我們的應(yīng)用層 。

第二個(gè)子接口 ChannelOutboundHandler 是處理寫(xiě)數(shù)據(jù)的邏輯 ， 他是定義我們一段在組裝完響應(yīng)之后吧數(shù)據(jù)寫(xiě)到對(duì)端的邏輯 ，比如我們封裝好一個(gè)response對(duì)象 ， 接下來(lái)我們可能對(duì)這個(gè)response做一些其他的特殊的邏輯， 然后 ， 在編碼成byteBuf ， 最終寫(xiě)到對(duì)端 ， 它里面最核心的一個(gè)方法就是write() ， 讀者可以將ChannelOutboundHandler的邏輯處理過(guò)程與TCP 的七層協(xié)議的封裝過(guò)程聯(lián)系起來(lái) ， 我們?cè)趹?yīng)用層組裝響應(yīng)之后 ， 通過(guò)層層協(xié)議的封裝 ， 直到最底層的物理層。

這兩個(gè)子接口分別有對(duì)應(yīng)的默認(rèn)的實(shí)現(xiàn) ， ChannelInboundHandlerAdapter 和ChannelOutboundHandlerAdapter ， 他們分別實(shí)現(xiàn)了兩大接口的所有功能 ， 默認(rèn)情況下回吧讀寫(xiě)事件傳播到下一個(gè)handler。

說(shuō)了這么多理論， 其實(shí)還是比較抽象的 ， 下面我們就用一個(gè)具體的demo 來(lái)學(xué)習(xí)一下這兩個(gè)handler的事件傳播機(jī)制。

四、 ChannelInboundHanndler 的事件傳播

關(guān)于ChannelInboundHandler , 我們拿 ChannelRead() 為例子 ， 來(lái)體驗(yàn)一下inbound時(shí)間的傳播

我們?cè)诜?wù)端的pipeline 添加三個(gè)ChannelInboundHandler

Test_12_Server.javaserverBootstrap.group(bossGroup, workGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new Test_11_InboundHandlerA());ch.pipeline().addLast(new Test_11_InboundHandlerB());ch.pipeline().addLast(new Test_11_InboundHandlerC());}});

每個(gè)inBoundHandler 繼承自ChannelInboundHandlerAdapter ， 然后實(shí)現(xiàn)了 channelRead() 方法

/*** 2019年1月29日* @author outman* 服務(wù)端處理 A*/class Test_11_InboundHandlerA extends ChannelInboundHandlerAdapter{@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buffer = (ByteBuf)msg;System.out.println("Test_11_InboundHandlerA --> " + buffer.toString(Charset.forName("UTF-8")));super.channelRead(ctx, msg);}}/*** 2019年1月29日* @author outman* 服務(wù)端處理 B*/class Test_11_InboundHandlerB extends ChannelInboundHandlerAdapter{@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buffer = (ByteBuf)msg;System.out.println("Test_11_InboundHandlerB --> " + buffer.toString(Charset.forName("UTF-8")));super.channelRead(ctx, msg);}}/*** 2019年1月29日* @author outman* 服務(wù)端處理 C*/class Test_11_InboundHandlerC extends ChannelInboundHandlerAdapter{@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buffer = (ByteBuf)msg;System.out.println("Test_11_InboundHandlerC --> " + buffer.toString(Charset.forName("UTF-8")));super.channelRead(ctx, msg);}}

在channelRead() 方法里面 ， 我們當(dāng)前handler 的信息 ， 然后調(diào)用父類(lèi)的channelRead() 方法 ， 而這里父類(lèi)的channelRead()方法會(huì)自定調(diào)用下一個(gè)inboundHandler的channelRead() , 并且會(huì)把處理完畢的對(duì)象傳遞給下一個(gè)inboundHandler , 我們例子中傳遞的對(duì)象都是同一個(gè)msg 。

我們通過(guò) addLast()方法來(lái)為pipeline添加 inboundHandler , 當(dāng)然 ， 除了這個(gè)方法還有其他的方法 ， 感興趣的同學(xué)可以去瀏覽一下pipeline的api ，這里我們添加的順序?yàn)?A --> B --> c , 然后我們來(lái)看一下控制臺(tái)輸出

五、ChannelOutboundHandler 的時(shí)間傳播

關(guān)于ChannelOutboundHandler , 我們拿write()為例子，來(lái)體驗(yàn)一下outbound事件的傳播。

我們繼續(xù)在服務(wù)端的pipeline添加三個(gè)ChannelOutboundHandler

Test_12_Server.javaserverBootstrap.group(bossGroup, workGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {// inbound 服務(wù)端讀數(shù)據(jù)邏輯ch.pipeline().addLast(new Test_11_InboundHandlerA());ch.pipeline().addLast(new Test_11_InboundHandlerB());ch.pipeline().addLast(new Test_11_InboundHandlerC());// outbound 服務(wù)端系數(shù)據(jù)邏輯ch.pipeline().addLast(new Test_11_OutboundHandlerA());ch.pipeline().addLast(new Test_11_OutboundHandlerB());ch.pipeline().addLast(new Test_11_OutboundHandlerC());}});

每個(gè)outboundHandler都繼承自ChannelOutboundHandlerAdapter , 然后實(shí)現(xiàn)了write()方法

/*** 2019年2月14日* @author outman** 服務(wù)端寫(xiě)數(shù)據(jù)邏輯 A*/class Test_12_OutboundHandlerA extends ChannelOutboundHandlerAdapter {@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {System.out.println("Test_12_OutboundHandlerA --> " + msg);super.write(ctx, msg, promise);}}/*** 2019年2月14日* @author outman** 服務(wù)端寫(xiě)數(shù)據(jù)邏輯 B*/class Test_12_OutboundHandlerB extends ChannelOutboundHandlerAdapter {@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {System.out.println("Test_12_OutboundHandlerB --> " + msg);super.write(ctx, msg, promise);}}/*** 2019年2月14日* @author outman** 服務(wù)端寫(xiě)數(shù)據(jù)邏輯 C*/class Test_12_OutboundHandlerC extends ChannelOutboundHandlerAdapter {@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {System.out.println("Test_12_OutboundHandlerC --> " + msg);super.write(ctx, msg, promise);}}

在write()方法里面 ， 我們打印當(dāng)前handler的信息，然后調(diào)用父類(lèi)write()方法，而這里父類(lèi)的write()方法會(huì)自動(dòng)調(diào)用到下一個(gè)outboundHandler的write()方法，并且把當(dāng)前outboundHandler里處理完畢的對(duì)象傳遞到下一個(gè)outboundHandler.

我們通過(guò)addLast()方法添加outboundHandler的順序?yàn)?A -> B -> C , 最后我們來(lái)看一下控制臺(tái)的輸出

可以看到outboundHandler的執(zhí)行順序與我們添加的順序相反 ， 最后，我們?cè)趤?lái)看一下pipeline的結(jié)構(gòu)和執(zhí)行順序。

pipeline 的結(jié)構(gòu) 不管我們定義的是那種類(lèi)型的handler， 最終他們都以雙向鏈表的形式連接，這里實(shí)際鏈表的節(jié)點(diǎn)是ChannelHandlerContext ， 這里為了讓結(jié)構(gòu)清晰突出，可以直接把節(jié)點(diǎn)看做ChannelHandlerContext

pipeline 的執(zhí)行順序 雖然兩種類(lèi)型的handler在一個(gè)雙向鏈表里， 但是這兩類(lèi)handler 的分工是不一樣的，inboundHandler的事件通常只會(huì)傳播到下一個(gè)channelHandler,outboundHandler的事件通常通常只會(huì)傳播到下一個(gè)outboundHandler ， 兩者互不干擾。

六 、 總結(jié)

通過(guò)我們前面編寫(xiě)客戶(hù)端、服務(wù)端處理邏輯引出了pipeline和channelHandler的概念

channelHandler分為inbound和outbound兩種類(lèi)型的接口 ， 分別是處理數(shù)據(jù)讀和數(shù)據(jù)寫(xiě)的邏輯

兩種類(lèi)型的handler均有相應(yīng)的默認(rèn)實(shí)現(xiàn)，默認(rèn)會(huì)把事件傳遞到下一個(gè) ， 這里的傳遞事件其實(shí)說(shuō)白了就是把本handler的處理結(jié)果傳遞給下一個(gè)handler繼續(xù)處理

inboundHandler的執(zhí)行順序與我們實(shí)際的添加順序相同 ，而outboundHandler 相反。

七 、 思考

參考本文中的例子 ， 如果我們往pipeline里面添加handler的順序不變，要在控制臺(tái)打印出inboundA -> inboundB -> outboundB -> outboundA , 該如何實(shí)現(xiàn)？

如何在每個(gè)handler里面打印上一個(gè)handler 的處理結(jié)束的時(shí)間點(diǎn)呢？

答： 可以將上一個(gè)handler處理結(jié)束的時(shí)間放在channel的attr中

如何讓outboundHandler按照添加順序執(zhí)行？

答： pipeline有andLast() 和andFrist()方法 ，inboundHandler在執(zhí)行過(guò)程中正向遍歷鏈表 使用andLast()方法先進(jìn)先出可以順序執(zhí)行 ， outboundHandler在執(zhí)行過(guò)程中逆向遍歷鏈表， 使用andFrist()方法 先進(jìn)后出可以順序執(zhí)行 。

OutBoundHandler一直沒(méi)有被執(zhí)行到，pipeline也已添加，有可能是什么原因呢？

答： ctx.channel().writeAndFlush(msg); 進(jìn)行寫(xiě)事件才會(huì)觸發(fā)out調(diào)用鏈 。 所以需要將inboundHandlerC 中執(zhí)行寫(xiě)邏輯 才會(huì)執(zhí)行outboundHandler.

總結(jié)

以上是生活随笔為你收集整理的Netty实战 IM即时通讯系统（十一）pipeline与channelHandler的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。