本篇文章主要詳細分析Netty中的核心組件，
啟動器Bootstrap和ServerBootstrap作為Netty構建客戶端和服務端的路口，是撰寫Netty網路程式的第一步，它可以讓我們把Netty的核心組件像搭積木一樣組裝在一起，在Netty Server端構建的程序中，我們需要關注三個重要的步驟
- 配置執行緒池
- Channel初始化
- Handler處理器構建
調度器詳解
前面我們講過NIO多路復用的設計模式之Reactor模型，Reactor模型的主要思想就是把網路連接、事件分發、任務處理的職責進行分離，并且通過引入多執行緒來提高Reactor模型中的吞吐量，其中包括三種Reactor模型
- 單執行緒單Reactor模型
- 多執行緒單Reactor模型
- 多執行緒多Reactor模型
在Netty中，可以非常輕松的實作上述三種執行緒模型，并且Netty推薦使用主從多執行緒模型，這樣就可以輕松的實作成千上萬的客戶端連接的處理，在海量的客戶端并發請求中，主從多執行緒模型可以通過增加SubReactor執行緒數量，充分利用多核能力提升系統吞吐量，
Reactor模型的運行機制分為四個步驟，如圖2-10所示，
- 連接注冊，Channel建立后，注冊到Reactor執行緒中的Selector選擇器
- 事件輪詢，輪詢Selector選擇器中已經注冊的所有Channel的I/O事件
- 事件分發，為準備就緒的I/O事件分配相應的處理執行緒
- 任務處理，Reactor執行緒還負責任務佇列中的非I/O任務，每個Worker執行緒從各自維護的任務佇列中取出任務異步執行，
EventLoop事件回圈
在Netty中，Reactor模型的事件處理器是使用EventLoop來實作的，一個EventLoop對應一個執行緒，EventLoop內部維護了一個Selector和taskQueue，分別用來處理網路IO事件以及內部任務，它的作業原理如圖2-11所示，
EventLoop基本應用
下面這段代碼表示EventLoop，分別實作Selector注冊以及普通任務提交功能，
public class EventLoopExample {
public static void main(String[] args) {
EventLoopGroup group=new NioEventLoopGroup(2);
System.out.println(group.next()); //輸出第一個NioEventLoop
System.out.println(group.next()); //輸出第二個NioEventLoop
System.out.println(group.next()); //由于只有兩個，所以又會從第一個開始
//獲取一個事件回圈物件NioEventLoop
group.next().register(); //注冊到selector上
group.next().submit(()->{
System.out.println(Thread.currentThread().getName()+"-----");
});
}
}
EventLoop的核心流程
基于上述的講解，理解了EventLoop的作業機制后，我們再通過一個整體的流程圖來說明，如圖2-12所示，
EventLoop是一個Reactor模型的事件處理器，一個EventLoop對應一個執行緒，其內部會維護一個selector和taskQueue，負責處理IO事件和內部任務，IO事件和內部任務執行時間百分比通過ioRatio來調節，ioRatio表示執行IO時間所占百分比，任務包括普通任務和已經到時的延遲任務，延遲任務存放到一個優先級佇列PriorityQueue中，執行任務前從PriorityQueue讀取所有到時的task，然后添加到taskQueue中，最后統一執行task，
EventLoop如何實作多種Reactor模型
-
單執行緒模式
EventLoopGroup group=new NioEventLoopGroup(1); ServerBootstrap b=new ServerBootstrap(); b.group(group); -
多執行緒模式
EventLoopGroup group =new NioEventLoopGroup(); //默認會設定cpu核心數的2倍 ServerBootstrap b=new ServerBootstrap(); b.group(group); -
多執行緒主從模式
EventLoopGroup boss=new NioEventLoopGroup(1); EventLoopGroup work=new NioEventLoopGroup(); ServerBootstrap b=new ServerBootstrap(); b.group(boss,work);
EventLoop實作原理
-
EventLoopGroup初始化方法，在MultithreadEventExecutorGroup.java中，根據配置的nThreads數量，構建一個EventExecutor陣列
protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) { checkPositive(nThreads, "nThreads"); if (executor == null) { executor = new ThreadPerTaskExecutor(newDefaultThreadFactory()); } children = new EventExecutor[nThreads]; for (int i = 0; i < nThreads; i ++) { boolean success = false; try { children[i] = newChild(executor, args); } } } -
注冊channel到多路復用器的實作，MultithreadEventLoopGroup.register方法（）
SingleThreadEventLoop ->AbstractUnsafe.register ->AbstractChannel.register0->AbstractNioChannel.doRegister()
可以看到會把channel注冊到某一個eventLoop中的unwrappedSelector復路器中，
protected void doRegister() throws Exception { boolean selected = false; for (;;) { try { selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this); return; } } } -
事件處理程序，通過NioEventLoop中的run方法不斷遍歷
protected void run() { int selectCnt = 0; for (;;) { try { int strategy; try { //計算策略，根據阻塞佇列中是否含有任務來決定當前的處理方式 strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks()); switch (strategy) { case SelectStrategy.CONTINUE: continue; case SelectStrategy.BUSY_WAIT: // fall-through to SELECT since the busy-wait is not supported with NIO case SelectStrategy.SELECT: long curDeadlineNanos = nextScheduledTaskDeadlineNanos(); if (curDeadlineNanos == -1L) { curDeadlineNanos = NONE; // nothing on the calendar } nextWakeupNanos.set(curDeadlineNanos); try { if (!hasTasks()) { //如果佇列中資料為空，則呼叫select查詢就緒事件 strategy = select(curDeadlineNanos); } } finally { nextWakeupNanos.lazySet(AWAKE); } default: } } selectCnt++; cancelledKeys = 0; needsToSelectAgain = false; /* ioRatio調節連接事件和內部任務執行事件百分比 * ioRatio越大，連接事件處理占用百分比越大 */ final int ioRatio = this.ioRatio; boolean ranTasks; if (ioRatio == 100) { try { if (strategy > 0) { //處理IO時間 processSelectedKeys(); } } finally { //確保每次都要執行佇列中的任務 ranTasks = runAllTasks(); } } else if (strategy > 0) { final long ioStartTime = System.nanoTime(); try { processSelectedKeys(); } finally { // Ensure we always run tasks. final long ioTime = System.nanoTime() - ioStartTime; ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio); } } else { ranTasks = runAllTasks(0); // This will run the minimum number of tasks } if (ranTasks || strategy > 0) { if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS && logger.isDebugEnabled()) { logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.", selectCnt - 1, selector); } selectCnt = 0; } else if (unexpectedSelectorWakeup(selectCnt)) { // Unexpected wakeup (unusual case) selectCnt = 0; } } }
服務編排層Pipeline的協調處理
通過EventLoop可以實作任務的調度，負責監聽I/O事件、信號事件等，當收到相關事件后，需要有人來回應這些事件和資料，而這些事件是通過ChannelPipeline中所定義的ChannelHandler完成的，他們是Netty中服務編排層的核心組件，
在下面這段代碼中，我們增加了h1和h2兩個InboundHandler，用來處理客戶端資料的讀取操作，代碼如下，
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
//配置Server的通道，相當于NIO中的ServerSocketChannel
.channel(NioServerSocketChannel.class)
//childHandler表示給worker那些執行緒配置了一個處理器，
// 這個就是上面NIO中說的，把處理業務的具體邏輯抽象出來，放到Handler里面
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
// socketChannel.pipeline().addLast(new NormalMessageHandler());
socketChannel.pipeline().addLast("h1",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("handler-01");
super.channelRead(ctx, msg);
}
}).addLast("h2",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("handler-02");
super.channelRead(ctx, msg);
}
});
}
});
上述代碼構建了一個ChannelPipeline，得到如圖2-13所示的結構，每個Channel都會系結一個ChannelPipeline，一個ChannelPipeline包含多個ChannelHandler，這些Handler會被包裝成ChannelHandlerContext加入到Pipeline構建的雙向鏈表中，
ChannelHandlerContext用來保存ChannelHandler的背景關系，它包含了ChannelHandler生命周期中的所有事件，比如connect/bind/read/write等，這樣設計的好處是，各個ChannelHandler進行資料傳遞時，前置和后置的通用邏輯就可以直接保存到ChannelHandlerContext中進行傳遞，
出站和入站操作
根據網路資料的流向，ChannelPipeline分為入站ChannelInBoundHandler和出站ChannelOutboundHandler兩個處理器，如圖2-14所示，客戶端與服務端通信程序中，資料從客戶端發向服務端的程序叫出站，對于服務端來說，資料從客戶端流入到服務端，這個時候是入站，
ChannelHandler事件觸發機制
當某個Channel觸發了IO事件后，會通過Handler進行處理，而ChannelHandler是圍繞I/O事件的生命周期來設計的，比如建立連接、讀資料、寫資料、連接銷毀等，
ChannelHandler有兩個重要的子介面實作，分別攔截資料流入和資料流出的I/O事件
- ChannelInboundHandler
- ChannelOutboundHandler
圖2-15中顯示的Adapter類，提供很多默認操作，比如ChannelHandler中有很多很多方法，我們用戶自定義的方法有時候不需要多載全部，只需要多載一兩個方法，那么可以使用Adapter類，它里面有很多默認的方法，其它框架中結尾是Adapter的類的作用也大都是如此，所以我們在使用netty的時候，往往很少直接實作ChannelHandler的介面，經常是繼承Adapter類，
ChannelInboundHandler事件回呼和觸發時機如下
| 事件回呼方法 | 觸發時機 |
|---|---|
| channelRegistered | Channel 被注冊到 EventLoop |
| channelUnregistered | Channel 從 EventLoop 中取消注冊 |
| channelActive | Channel 處于就緒狀態，可以被讀寫 |
| channelInactive | Channel 處于非就緒狀態 |
| channelRead | Channel 可以從遠端讀取到資料 |
| channelReadComplete | Channel 讀取資料完成 |
| userEventTriggered | 用戶事件觸發時 |
| channelWritabilityChanged | Channel 的寫狀態發生變化 |
ChannelOutboundHandler時間回呼觸發時機
| 事件回呼方法 | 觸發時機 |
|---|---|
| bind | 當請求將channel系結到本地地址時被呼叫 |
| connect | 當請求將channel連接到遠程節點時被呼叫 |
| disconnect | 當請求將channel從遠程節點斷開時被呼叫 |
| close | 當請求關閉channel時被呼叫 |
| deregister | 當請求將channel從它的EventLoop注銷時被呼叫 |
| read | 當請求通過channel讀取資料時被呼叫 |
| flush | 當請求通過channel將入隊資料重繪到遠程節點時呼叫 |
| write | 當請求通過channel將資料寫到遠程節點時被呼叫 |
事件傳播機制演示
public class NormalOutBoundHandler extends ChannelOutboundHandlerAdapter {
private final String name;
public NormalOutBoundHandler(String name) {
this.name = name;
}
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
System.out.println("OutBoundHandler:"+name);
super.write(ctx, msg, promise);
}
}
public class NormalInBoundHandler extends ChannelInboundHandlerAdapter {
private final String name;
private final boolean flush;
public NormalInBoundHandler(String name, boolean flush) {
this.name = name;
this.flush = flush;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("InboundHandler:"+name);
if(flush){
ctx.channel().writeAndFlush(msg);
}else {
super.channelRead(ctx, msg);
}
}
}
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
//配置Server的通道，相當于NIO中的ServerSocketChannel
.channel(NioServerSocketChannel.class)
//childHandler表示給worker那些執行緒配置了一個處理器，
// 這個就是上面NIO中說的，把處理業務的具體邏輯抽象出來，放到Handler里面
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline()
.addLast(new NormalInBoundHandler("NormalInBoundA",false))
.addLast(new NormalInBoundHandler("NormalInBoundB",false))
.addLast(new NormalInBoundHandler("NormalInBoundC",true));
socketChannel.pipeline()
.addLast(new NormalOutBoundHandler("NormalOutBoundA"))
.addLast(new NormalOutBoundHandler("NormalOutBoundB"))
.addLast(new NormalOutBoundHandler("NormalOutBoundC"));
}
});
上述代碼運行后會得到如下執行結果
InboundHandler:NormalInBoundA
InboundHandler:NormalInBoundB
InboundHandler:NormalInBoundC
OutBoundHandler:NormalOutBoundC
OutBoundHandler:NormalOutBoundB
OutBoundHandler:NormalOutBoundA
當客戶端向服務端發送請求時，會觸發服務端的NormalInBound呼叫鏈，按照排列順序逐個呼叫Handler，當InBound處理完成后呼叫WriteAndFlush方法向客戶端寫回資料，此時會觸發NormalOutBoundHandler呼叫鏈的write事件，
從執行結果來看，Inbound和Outbound的事件傳播方向是不同的，Inbound傳播方向是head->tail，Outbound傳播方向是Tail-Head，
例外傳播機制
ChannelPipeline時間傳播機制是典型的責任鏈模式，那么有同學肯定會有疑問，如果這條鏈路中某個handler出現例外，那會導致什么問題呢？我們對前面的例子修改NormalInBoundHandler
public class NormalInBoundHandler extends ChannelInboundHandlerAdapter {
private final String name;
private final boolean flush;
public NormalInBoundHandler(String name, boolean flush) {
this.name = name;
this.flush = flush;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("InboundHandler:"+name);
if(flush){
ctx.channel().writeAndFlush(msg);
}else {
//增加例外處理
throw new RuntimeException("InBoundHandler:"+name);
}
}
}
這個時候一旦拋出例外，會導致整個請求鏈被中斷，在ChannelHandler中提供了一個例外捕獲方法，這個方法可以避免ChannelHandler鏈中某個Handler例外導致請求鏈路中斷，它會把例外按照Handler鏈路的順序從head節點傳播到Tail節點，如果用戶最終沒有對例外進行處理，則最后由Tail節點進行統一處理
修改NormalInboundHandler，重寫下面這個方法，
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("InboundHandlerException:"+name);
super.exceptionCaught(ctx, cause);
}
在Netty應用開發中，好的例外處理非常重要能夠讓問題排查變得很輕松，所以我們可以通過一種統一攔截的方式來解決例外處理問題，
添加一個復合處理器實作類
public class ExceptionHandler extends ChannelDuplexHandler {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
if(cause instanceof RuntimeException){
System.out.println("處理業務例外");
}
super.exceptionCaught(ctx, cause);
}
}
把新增的ExceptionHandler添加到ChannelPipeline中
bootstrap.group(bossGroup, workerGroup)
//配置Server的通道，相當于NIO中的ServerSocketChannel
.channel(NioServerSocketChannel.class)
//childHandler表示給worker那些執行緒配置了一個處理器，
// 這個就是上面NIO中說的，把處理業務的具體邏輯抽象出來，放到Handler里面
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline()
.addLast(new NormalInBoundHandler("NormalInBoundA",false))
.addLast(new NormalInBoundHandler("NormalInBoundB",false))
.addLast(new NormalInBoundHandler("NormalInBoundC",true));
socketChannel.pipeline()
.addLast(new NormalOutBoundHandler("NormalOutBoundA"))
.addLast(new NormalOutBoundHandler("NormalOutBoundB"))
.addLast(new NormalOutBoundHandler("NormalOutBoundC"))
.addLast(new ExceptionHandler());
}
});
最終，我們就能夠實作例外的統一處理，
著作權宣告：本博客所有文章除特別宣告外，均采用 CC BY-NC-SA 4.0 許可協議，轉載請注明來自
Mic帶你學架構！
如果本篇文章對您有幫助，還請幫忙點個關注和贊，您的堅持是我不斷創作的動力，歡迎關注「跟著Mic學架構」公眾號公眾號獲取更多技術干貨！
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/354821.html
標籤：Java