由上小節(jié)可知,NIO 的類庫和 API 很繁雜,例如Selector、ServerSocketChannel、 SocketChannel、 ByteBuffer...這些對(duì)于從事應(yīng)用層的程序員來說,使用起來是比較麻煩的;另一方面,client面臨斷連重連、 網(wǎng)絡(luò)閃斷、心跳處理、半包讀寫、 網(wǎng)絡(luò)擁塞和異常流的處理等開發(fā)工作也是難度很大。
Netty 對(duì) JDK 自帶的 NIO 的 API 進(jìn)行了良好的封裝,解決了上述問題。例如,在Netty編程中,各種Channel里的存取數(shù)據(jù)的操作都被封裝好了,不需要涉及ByteBuffer等反人類設(shè)計(jì)的API。Netty都是模版式代碼,很多東西不用自己實(shí)現(xiàn),可以讓程序員更專注于業(yè)務(wù)邏輯,這些都會(huì)在本文及之后的文章中詳細(xì)體現(xiàn)。
此外,Netty還擁有高性能、高吞吐量、低延遲、最小化不必要的內(nèi)存復(fù)制等優(yōu)點(diǎn)。
在了解Netty使用場景后,本節(jié)將從IO模型的演進(jìn)角度來分析Netty線程模型,通過并發(fā)編程之父Doug Lea所寫《Scalable IO in Java》中涉及的一些IO處理模式,一步一步深入理解Netty線程模型的“進(jìn)化歷史”。在理解Netty線程模型后,介紹Netty模板式編程的通用組件,通過一個(gè)客戶端與服務(wù)端使用netty框架通信的事例,展示其核心功能。
Netty使用場景
1、互聯(lián)網(wǎng)行業(yè):在分布式系統(tǒng)中,各個(gè)節(jié)點(diǎn)之間需要遠(yuǎn)程服務(wù)調(diào)用,高性能的 RPC 框架必不可少,Netty 作為異步高性能的通信框架,往往作為基礎(chǔ)通信組件被這些 RPC 框架使用。典型的應(yīng)用有:阿里分布式服務(wù)框架 Dubbo 的 RPC 使用 Dubbo 協(xié)議進(jìn)行通信,Dubbo 協(xié)議默認(rèn)使用 Netty 作為基礎(chǔ)通信組件,用于實(shí)現(xiàn)各進(jìn)程節(jié)點(diǎn)之間的內(nèi)部通信;消息中間件Rocketmq底層也是用的Netty作為基礎(chǔ)通信組件。
2、游戲行業(yè):無論是手游服務(wù)端還是大型的網(wǎng)絡(luò)游戲,Java 語言都得到了越來越廣泛的應(yīng)用。Netty 作為高性能的基礎(chǔ)通信組件,它本身提供了 TCP/UDP 和 HTTP 協(xié)議棧。
3、大數(shù)據(jù)領(lǐng)域:經(jīng)典的 Hadoop 的高性能通信和序列化組件 Avro 的 RPC 框架,默認(rèn)采用 Netty 進(jìn)行跨節(jié)點(diǎn)通信,它的 Netty Service 是基于 Netty 框架二次封裝實(shí)現(xiàn)。
Netty相關(guān)開源項(xiàng)目:https://netty.io/wiki/related-projects.html
IO處理模式演進(jìn)
基本上所有的網(wǎng)絡(luò)處理程序都遵循以下基本的處理(handler)流程:
1、Read request (接收二進(jìn)制數(shù)據(jù))
2、Decode request (解碼為可讀數(shù)據(jù))
3、Process service (對(duì)數(shù)據(jù)進(jìn)行處理產(chǎn)生結(jié)果)
4、Encode reply (將結(jié)果編碼為二進(jìn)制數(shù)據(jù))
5、Send reply (返回結(jié)果)

BIO模型對(duì)于每一個(gè)請(qǐng)求都分發(fā)給一個(gè)線程(可以理解為一個(gè)handler),每個(gè)handler中都獨(dú)自處理上面1-5流程。這種模型的適用場景和瓶頸我們?cè)谏瞎?jié)已經(jīng)分析過了。
改進(jìn):采用基于事件驅(qū)動(dòng)的設(shè)計(jì),當(dāng)有事件觸發(fā)時(shí),才會(huì)調(diào)用處理器進(jìn)行數(shù)據(jù)處理(非阻塞)。這也就是我們上一節(jié)詳細(xì)分析過的NIO線程模型。

關(guān)于Reactor模式的兩個(gè)概念:
Reactor:負(fù)責(zé)響應(yīng)IO事件,當(dāng)檢測(cè)到一個(gè)新的事件,將其發(fā)送給相應(yīng)的Handler去處理。
Handler:負(fù)責(zé)處理非阻塞的行為,標(biāo)識(shí)系統(tǒng)管理的資源;同時(shí)將handler與事件綁定。
注意Reactor為單個(gè)線程,如上圖所示,不僅需要處理客戶端的accept連接請(qǐng)求,同時(shí)也負(fù)責(zé)分發(fā)(dispatch)讀寫等請(qǐng)求到處理器中。由于只有單個(gè)線程處理各種請(qǐng)求,所以要求處理器中的業(yè)務(wù)需要能夠快速處理完。
改進(jìn): 現(xiàn)在的服務(wù)器基本是多核CPU,那么在多處理器場景下,為實(shí)現(xiàn)服務(wù)的高性能我們可以有目的的采用多線程模式處理業(yè)務(wù):

這種模型下,客戶端來的連接和注冊(cè)還是由主線程Reactor統(tǒng)一去處理,只不過客戶端連接成功后的后續(xù)事件分發(fā)給worker線程池去處理而已。
但是,當(dāng)客戶端短時(shí)間幾十萬條連接來的時(shí)候(雙11,春運(yùn)搶票),單個(gè)Reator不僅要處理注冊(cè)事件,也要同時(shí)分發(fā)任務(wù)到Worker線程池,由于分發(fā)也是比較耗時(shí)的操作,從而有可能導(dǎo)致阻塞。
繼續(xù)改進(jìn):Doug Lea老同志又將Reactor拆分為兩部分:

通俗理解,mainReactor就是大總管,只負(fù)責(zé)接客,subReactor就是一個(gè)員工,負(fù)責(zé)給總管接來的客戶提供服務(wù)。
Netty線程模型就是由這個(gè)主從NIO模型演變而成的。
Netty線程模型
如下圖所示,Boss Group就是上圖里的mainReactor,與上圖不同在于Worker Group ,它可以理解為一組subReactor,即大總管下面有多個(gè)員工來干活,每次接的客戶都均勻分配給不同員工。netty之所以單機(jī)支持百萬級(jí)別并發(fā)量,就是因?yàn)?strong>一主多從的線程模型。

需要說明的是,Netty 的線程模型并不是一成不變的。它通常采用一主多從,但是也可以根據(jù)實(shí)際需要配置啟動(dòng)參數(shù),通過設(shè)置不同的啟動(dòng)參數(shù),Netty 可以同時(shí)支持“多主多從”。
下面是對(duì)上圖“一主多從”Netty模型的詳細(xì)解釋:
1、Netty 抽象出兩組線程池BossGroup和WorkerGroup,BossGroup(老板)專門負(fù)責(zé)接收客戶端的連接,WorkerGroup(員工們)專門負(fù)責(zé)網(wǎng)絡(luò)的讀寫。
2、 BossGroup和WorkerGroup類型都是NioEventLoopGroup。
3、NioEventLoopGroup 相當(dāng)于一個(gè)事件循環(huán)線程組, 這個(gè)組中含有多個(gè)事件循環(huán)線程 , 每一個(gè)事件循環(huán)線程是NioEventLoop。
4、每個(gè)NioEventLoop都有一個(gè)selector , 用于監(jiān)聽注冊(cè)在其上的socketChannel的網(wǎng)絡(luò)通訊。
5、每個(gè)Boss NioEventLoop線程內(nèi)部循環(huán)執(zhí)行的步驟有 3 步:
a:處理accept事件 , 與client 建立連接 , 生成 NioSocketChannel 。
b: 將NioSocketChannel注冊(cè)到某個(gè)worker NIOEventLoop上的selector。
c:處理任務(wù)隊(duì)列的任務(wù) , 即runAllTasks。
6、每個(gè)worker NIOEventLoop線程循環(huán)執(zhí)行的步驟:
a:輪詢注冊(cè)到自己selector上的所有NioSocketChannel 的read, write事件。
b:處理 I/O 事件, 即read , write 事件, 在對(duì)應(yīng)NioSocketChannel 處理業(yè)務(wù)。
c:runAllTasks處理任務(wù)隊(duì)列TaskQueue的任務(wù) ,一些耗時(shí)的業(yè)務(wù)處理一般可以放入TaskQueue中慢慢處理,這樣不影響數(shù)據(jù)在 pipeline 中的流動(dòng)處理。
7、每個(gè)worker NIOEventLoop處理NioSocketChannel業(yè)務(wù)時(shí),會(huì)使用 pipeline (管道),管道中維護(hù)了很多 handler 處理器用來處理 channel 中的數(shù)據(jù)。
Netty模塊組件
Netty是模版式編程,原理不易但是使用簡單,下面介紹一下Netty模板中通用的組件。
【Bootstrap、ServerBootstrap】
Bootstrap 意思是引導(dǎo),一個(gè) Netty 應(yīng)用通常由一個(gè) Bootstrap 開始,主要作用是配置整個(gè) Netty程序,通過鏈?zhǔn)秸{(diào)用串聯(lián)各個(gè)組件。Netty 中 Bootstrap 類是客戶端程序的啟動(dòng)引導(dǎo)類,ServerBootstrap 是服務(wù)端 啟動(dòng)引導(dǎo)類。
【Future、ChannelFuture】:
正如前面介紹,在 Netty 中所有的 IO 操作都是異步的,不能立刻得知消息是否被正確處理。但是可以過一會(huì)等它執(zhí)行完成或者直接注冊(cè)一個(gè)監(jiān)聽,具體的實(shí)現(xiàn)就是通過 Future 和 ChannelFutures,他們可以注冊(cè)一個(gè)監(jiān)聽,當(dāng)操作執(zhí)行成功或失敗時(shí)監(jiān)聽會(huì)自動(dòng)觸發(fā)注冊(cè)的監(jiān)聽事件。
【Channel】:
Netty 網(wǎng)絡(luò)通信的組件,能夠用于執(zhí)行網(wǎng)絡(luò) I/O 操作。Channel 為用戶提供:
1、當(dāng)前網(wǎng)絡(luò)連接的通道的狀態(tài)(例如是否打開?是否已連接?)
2、網(wǎng)絡(luò)連接的配置參數(shù) (例如接收緩沖區(qū)大?。?br>
3、提供異步的網(wǎng)絡(luò) I/O 操作(如建立連接,讀寫,綁定端口),異步調(diào)用意味著任何 I/O 調(diào)用都將立即返回,并且不保證在調(diào)用結(jié)束時(shí)所請(qǐng)求的 I/O 操作已完成。
4、調(diào)用立即返回一個(gè) ChannelFuture 實(shí)例,通過注冊(cè)監(jiān)聽器到 ChannelFuture 上,可以 I/O 操作成功、失敗或取消時(shí)回調(diào)通知調(diào)用方。
5、支持關(guān)聯(lián) I/O 操作與對(duì)應(yīng)的處理程序。不同協(xié)議、不同的阻塞類型的連接都有不同的Channel 類型與之對(duì)應(yīng)。下面是一些常用的 Channel 類型:
1、NioSocketChannel,異步的客戶端 TCP Socket 連接。(最常用)
2、NioServerSocketChannel,異步的服務(wù)器端 TCP Socket 連接
3、NioDatagramChannel,異步的 UDP 連接
4、NioSctpChannel,異步的客戶端 Sctp 連接
5、NioSctpServerChannel,異步的 Sctp 服務(wù)器端連接,這些通道涵蓋了 UDP 和 TCP 網(wǎng)絡(luò) IO 以及文件 IO。
【Selector】:
Netty 基于 Selector 對(duì)象實(shí)現(xiàn) I/O 多路復(fù)用,通過 Selector 一個(gè)線程可以監(jiān)聽多個(gè)連接的 Channel 事件。當(dāng)向一個(gè) Selector 中注冊(cè) Channel 后,Selector 內(nèi)部的機(jī)制就可以自動(dòng)不斷地查詢(Select) 這些注冊(cè)的 Channel 是否有已就緒的 I/O 事件(例如可讀,可寫,網(wǎng)絡(luò)連接完成等),這樣程序就可以很簡單地使用一個(gè)線程高效地管理多個(gè) Channel 。
【NioEventLoop】:
NioEventLoop 中維護(hù)了一個(gè)線程和任務(wù)隊(duì)列,支持異步提交執(zhí)行任務(wù),線程啟動(dòng)時(shí)會(huì)調(diào)用 NioEventLoop 的 run 方法,執(zhí)行 I/O 任務(wù)和非 I/O 任務(wù):
1、I/O 任務(wù)即 selectionKey 中 ready 的事件,如 accept、connect、read、write 等,由 processSelectedKeys 方法觸發(fā)。
2、非 IO 任務(wù),添加到 taskQueue 中的任務(wù),如 register0、bind0 等任務(wù),由 runAllTasks 方法觸發(fā)。
【NioEventLoopGroup】:NioEventLoopGroup,主要管理 eventLoop 的生命周期,可以理解為一個(gè)線程池,內(nèi)部維護(hù)了一組線程,每個(gè)線程(NioEventLoop)負(fù)責(zé)處理多個(gè) Channel 上的事件,而一個(gè) Channel 只對(duì)應(yīng)于一個(gè)線程。
【ChannelHandler】:ChannelHandler 是一個(gè)接口,處理 I/O 事件或攔截 I/O 操作,并將其轉(zhuǎn)發(fā)到其 ChannelPipeline(業(yè)務(wù)處理鏈)中的下一個(gè)處理程序。ChannelHandler 本身并沒有提供很多方法,因?yàn)檫@個(gè)接口有許多的方法需要實(shí)現(xiàn),方便使用期間,可以繼承它的子類:
ChannelInboundHandler 用于處理入站 I/O 事件
ChannelOutboundHandler 用于處理出站 I/O 操作
或者使用以下適配器類:
ChannelInboundHandlerAdapter 用于處理入站 I/O 事件。
ChannelOutboundHandlerAdapter 用于處理出站 I/O 操作。
【ChannelHandlerContext】:
保存 Channel 相關(guān)的所有上下文信息,同時(shí)關(guān)聯(lián)一個(gè) ChannelHandler 對(duì)象。
【ChannelPipline】:
保存 ChannelHandler 的 List,用于處理或攔截 Channel 的入站事件和出站操作。ChannelPipeline 實(shí)現(xiàn)了一種高級(jí)形式的攔截過濾器模式,使用戶可以完全控制事件的處理方式,以及 Channel 中各個(gè)的 ChannelHandler 如何相互交互。在 Netty 中每個(gè) Channel 都有且僅有一個(gè) ChannelPipeline 與之對(duì)應(yīng),它們的組成關(guān)系如下:

一個(gè) Channel 包含了一個(gè) ChannelPipeline,而 ChannelPipeline 中又維護(hù)了一個(gè)由 ChannelHandlerContext 組成的雙向鏈表,并且每個(gè) ChannelHandlerContext 中又關(guān)聯(lián)著一個(gè) ChannelHandler。read事件(入站事件)和write事件(出站事件)在一個(gè)雙向鏈表中,入站事件會(huì)從鏈表 head 往后傳遞到最后一個(gè)入站的 handler,出站事件會(huì)從鏈表 tail 往前傳遞到最前一個(gè)出站的 handler,兩種類型的 handler 互不干擾。
Netty通信事例
在熟悉了Netty模塊組件后,我們介紹一個(gè)Netty通信事例,代碼中用到了我們上文介紹的各種核心模塊組件。
服務(wù)端:
public class NettyServer {
public static void main(String[] args) throws Exception {
//創(chuàng)建兩個(gè)線程組bossGroup和workerGroup, 含有的子線程N(yùn)ioEventLoop的個(gè)數(shù)默認(rèn)為cpu核數(shù)的兩倍
// bossGroup只是處理連接請(qǐng)求 ,真正的和客戶端業(yè)務(wù)處理,會(huì)交給workerGroup完成
EventLoopGroup bossGroup = new NioEventLoopGroup(10);
EventLoopGroup workerGroup = new NioEventLoopGroup(100000);
try {
//創(chuàng)建服務(wù)器端的啟動(dòng)對(duì)象
ServerBootstrap bootstrap = new ServerBootstrap();
//使用鏈?zhǔn)骄幊虂砼渲脜?shù)
bootstrap.group(bossGroup, workerGroup) //設(shè)置兩個(gè)線程組
.channel(NioServerSocketChannel.class) //使用NioServerSocketChannel作為服務(wù)器的通道實(shí)現(xiàn)
// 初始化服務(wù)器連接隊(duì)列大小,服務(wù)端處理客戶端連接請(qǐng)求是順序處理的,所以同一時(shí)間只能處理一個(gè)客戶端連接。
// 多個(gè)客戶端同時(shí)來的時(shí)候,服務(wù)端將不能處理的客戶端連接請(qǐng)求放在隊(duì)列中等待處理
.option(ChannelOption.SO_BACKLOG, 1024)
.childHandler(new ChannelInitializer<SocketChannel>() {//創(chuàng)建通道初始化對(duì)象,設(shè)置初始化參數(shù)
@Override
protected void initChannel(SocketChannel ch) throws Exception {
//對(duì)workerGroup的SocketChannel設(shè)置處理器
ch.pipeline().addLast(new NettyServerHandler());
}
});
System.out.println("netty server start。。");
//綁定一個(gè)端口并且同步, 生成了一個(gè)ChannelFuture異步對(duì)象,通過isDone()等方法可以判斷異步事件的執(zhí)行情況
//啟動(dòng)服務(wù)器(并綁定端口),bind是異步操作,sync方法是等待異步操作執(zhí)行完畢
ChannelFuture cf = bootstrap.bind(9000).sync();
//給cf注冊(cè)監(jiān)聽器,監(jiān)聽我們關(guān)心的事件
/*cf.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (cf.isSuccess()) {
System.out.println("監(jiān)聽端口9000成功");
} else {
System.out.println("監(jiān)聽端口9000失敗");
}
}
});*/
//對(duì)通道關(guān)閉進(jìn)行監(jiān)聽,closeFuture是異步操作,監(jiān)聽通道關(guān)閉
// 通過sync方法同步等待通道關(guān)閉處理完畢,這里會(huì)阻塞等待通道關(guān)閉完成
cf.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
服務(wù)端所注冊(cè)的自定義回調(diào)函數(shù) NettyServerHandler:
**
* 自定義Handler需要繼承netty規(guī)定好的某個(gè)HandlerAdapter(規(guī)范)
*/
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
/**
* 讀取客戶端發(fā)送的數(shù)據(jù)
*
* @param ctx 上下文對(duì)象, 含有通道channel,管道pipeline
* @param msg 就是客戶端發(fā)送的數(shù)據(jù)
* @throws Exception
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("服務(wù)器讀取線程 " + Thread.currentThread().getName());
//Channel channel = ctx.channel();
//ChannelPipeline pipeline = ctx.pipeline(); //本質(zhì)是一個(gè)雙向鏈接, 出站入站
//將 msg 轉(zhuǎn)成一個(gè) ByteBuf,類似NIO 的 ByteBuffer
ByteBuf buf = (ByteBuf) msg;
System.out.println("客戶端發(fā)送消息是:" + buf.toString(CharsetUtil.UTF_8));
}
/**
* 數(shù)據(jù)讀取完畢處理方法
*
* @param ctx
* @throws Exception
*/
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ByteBuf buf = Unpooled.copiedBuffer("HelloClient".getBytes(CharsetUtil.UTF_8));
ctx.writeAndFlush(buf);
}
/**
* 處理異常, 一般是需要關(guān)閉通道
*
* @param ctx
* @param cause
* @throws Exception
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
}
這個(gè)類繼承了ChannelInboundHandlerAdapter的幾個(gè)方法:
channelRead方法:當(dāng)客戶端與服務(wù)端連通好之后,客戶端發(fā)數(shù)據(jù)時(shí),服務(wù)端會(huì)主動(dòng)調(diào)用這個(gè)方法。
channelReadComplete方法:數(shù)據(jù)處理完畢的方法,ctx.writeAndFlush()就可以往客戶端寫回?cái)?shù)據(jù)了。
客戶端:
public class NettyClient {
public static void main(String[] args) throws Exception {
//客戶端需要一個(gè)事件循環(huán)組
EventLoopGroup group = new NioEventLoopGroup();
try {
//創(chuàng)建客戶端啟動(dòng)對(duì)象
//注意客戶端使用的不是ServerBootstrap而是Bootstrap
Bootstrap bootstrap = new Bootstrap();
//設(shè)置相關(guān)參數(shù)
bootstrap.group(group) //設(shè)置線程組
.channel(NioSocketChannel.class) // 使用NioSocketChannel作為客戶端的通道實(shí)現(xiàn)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
//加入處理器
ch.pipeline().addLast(new NettyClientHandler());
}
});
System.out.println("netty client start。。");
//啟動(dòng)客戶端去連接服務(wù)器端
ChannelFuture cf = bootstrap.connect("127.0.0.1", 9000).sync();
//對(duì)通道關(guān)閉進(jìn)行監(jiān)聽
cf.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
}
客戶端自定義回調(diào)函數(shù):
public class NettyClientHandler extends ChannelInboundHandlerAdapter {
/**
* 當(dāng)客戶端連接服務(wù)器完成就會(huì)觸發(fā)該方法
*
* @param ctx
* @throws Exception
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ByteBuf buf = Unpooled.copiedBuffer("HelloServer".getBytes(CharsetUtil.UTF_8));
ctx.writeAndFlush(buf);
}
//當(dāng)通道有讀取事件時(shí)會(huì)觸發(fā),即服務(wù)端發(fā)送數(shù)據(jù)給客戶端
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
System.out.println("收到服務(wù)端的消息:" + buf.toString(CharsetUtil.UTF_8));
System.out.println("服務(wù)端的地址: " + ctx.channel().remoteAddress());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
與NettyServerHandler類似,其中的channelActive()方法是當(dāng)客戶端與服務(wù)器連接完成時(shí)候就會(huì)執(zhí)行的方法。
看完代碼,我們發(fā)現(xiàn)Netty框架的目標(biāo)就是讓開發(fā)者的業(yè)務(wù)邏輯從網(wǎng)絡(luò)基礎(chǔ)應(yīng)用編碼中分離出來,讓開發(fā)者可以專注業(yè)務(wù)的開發(fā),而不需寫一大堆類似NIO的網(wǎng)絡(luò)處理操作。在Netty系列之后的文章我們將進(jìn)一步體會(huì)到,開發(fā)者往往只需要寫類似于NettyServerHandler 、NettyClientHandler 等各種和業(yè)務(wù)有關(guān)的handler,將其加入主程序的pipeline即可,其他代碼基本上ctrl c+v 即可,是不是很簡單呢?
歡迎有興趣的同行關(guān)注。