基於大量圖片與實例深度解析Netty中的核心組件

本篇文章主要詳細分析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線程從各自維護的任務隊列中取出任務異步執行。

圖2-10 Reactor工作流程

EventLoop事件循環

在Netty中，Reactor模型的事件處理器是使用EventLoop來實現的，一個EventLoop對應一個線程，EventLoop內部維護了一個Selector和taskQueue，分別用來處理網絡IO事件以及內部任務，它的工作原理如圖2-11所示。

圖2-11 NioEventLoop原理

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。

圖2-12 EventLoop工作機制

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中進行傳遞。

圖2-13

出站和入站操作

根據網絡數據的流向，ChannelPipeline分為入站ChannelInBoundHandler和出站ChannelOutboundHandler兩個處理器，如圖2-14所示，客戶端與服務端通信過程中，數據從客戶端發向服務端的過程叫出站，對於服務端來說，數據從客戶端流入到服務端，這個時候是入站。

圖2-14 InBound和OutBound的關系

ChannelHandler事件觸發機制

當某個Channel觸發了IO事件后，會通過Handler進行處理，而ChannelHandler是圍繞I/O事件的生命周期來設計的，比如建立連接、讀數據、寫數據、連接銷毀等。

ChannelHandler有兩個重要的子接口實現，分別攔截數據流入和數據流出的I/O事件

• ChannelInboundHandler
• ChannelOutboundHandler

圖2-15中顯示的Adapter類，提供很多默認操作，比如ChannelHandler中有很多很多方法，我們用戶自定義的方法有時候不需要重載全部，只需要重載一兩個方法，那么可以使用Adapter類，它里面有很多默認的方法。其它框架中結尾是Adapter的類的作用也大都是如此。所以我們在使用netty的時候，往往很少直接實現ChannelHandler的接口，經常是繼承Adapter類。

圖2-15 ChannelHandler類關系圖

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學架構」公眾號公眾號獲取更多技術干貨！