Netty連接處理那些事

2019 年 11 月 12 日
筆記

編者註：Netty是Java領域有名的開源網路庫，特點是高性能和高擴展性，因此很多流行的框架都是基於它來構建的，比如我們熟知的Dubbo、Rocketmq、Hadoop等，針對高性能RPC，一般都是基於Netty來構建，比如soft-bolt。總之一句話，Java小夥伴們需要且有必要學會使用Netty並理解其實現原理。
關於Netty的入門講解可參考：Netty 入門，這一篇文章就夠了

Netty的連接處理就是IO事件的處理，IO事件包括讀事件、ACCEPT事件、寫事件和OP_CONNECT事件。

IO事件的處理是結合ChanelPipeline來做的，一個IO事件到來，首先進行數據的讀寫操作，然後交給ChannelPipeline進行後續處理，ChannelPipeline中包含了channelHandler鏈（head + 自定義channelHandler + tail）。
使用channelPipeline和channelHandler機制，起到了解耦和可擴展的作用。一個IO事件的處理，包含了多個處理流程，這些處理流程正好對應channelPipeline中的channelHandler。如果對數據處理有新的需求，那麼就新增channelHandler添加到channelPipeline中，這樣實現很6，以後自己寫程式碼可以參考。

說到這裡，一般為了滿足擴展性要求，常用2種模式：

方法模板模式：模板中定義了各個主流程，並且留下對應hook方法，便於擴展。
責任鏈模式：串列模式，可以動態添加鏈數量和對應回調方法。

netty的channelHandler的channelPipeline可以理解成就是責任鏈模式，通過動態增加channelHandler可達到復用和高擴展性目的。

了解netty連接處理機制之前需要了解下NioEventLoop模型，其中處理連接事件的架構圖如下：

對應的處理邏輯源碼為：

// 處理各種IO事件  private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {      final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();        try {          int readyOps = k.readyOps();          if ((readyOps & SelectionKey.OP_CONNECT) != 0) {              // OP_CONNECT事件，client連接上客戶端時觸發的事件              int ops = k.interestOps();              ops &= ~SelectionKey.OP_CONNECT;              k.interestOps(ops);              unsafe.finishConnect();          }            if ((readyOps & SelectionKey.OP_WRITE) != 0) {              ch.unsafe().forceFlush();          }            if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {              // 注意，這裡讀事件和ACCEPT事件對應的unsafe實例是不一樣的              // 讀事件 -> NioByteUnsafe,  ACCEPT事件 -> NioMessageUnsafe              unsafe.read();          }      } catch (CancelledKeyException ignored) {          unsafe.close(unsafe.voidPromise());      }  }

從上面程式碼來看，事件主要分為3種，分別是OP_CONNECT事件、寫事件和讀事件（也包括ACCEPT事件）。下面分為3部分展開：

ACCEPT事件

// NioMessageUnsafe  public void read() {      assert eventLoop().inEventLoop();      final ChannelConfig config = config();      final ChannelPipeline pipeline = pipeline();      final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();      allocHandle.reset(config);        boolean closed = false;      Throwable exception = null;      try {          do {              // 調用java socket的accept方法，接收請求              int localRead = doReadMessages(readBuf);              // 增加統計計數              allocHandle.incMessagesRead(localRead);          } while (allocHandle.continueReading());      } catch (Throwable t) {          exception = t;      }        // readBuf中存的是NioChannel      int size = readBuf.size();      for (int i = 0; i < size; i ++) {          readPending = false;          // 觸發fireChannelRead          pipeline.fireChannelRead(readBuf.get(i));      }      readBuf.clear();      allocHandle.readComplete();      pipeline.fireChannelReadComplete();  }

連接建立好之後就該連接的channel註冊到workGroup中某個NIOEventLoop的selector中，註冊操作是在fireChannelRead中完成的，這一塊邏輯就在ServerBootstrapAcceptor.channelRead中。

// ServerBootstrapAcceptor  public void channelRead(ChannelHandlerContext ctx, Object msg) {      final Channel child = (Channel) msg;        // 設置channel的pipeline handler，及channel屬性      child.pipeline().addLast(childHandler);      setChannelOptions(child, childOptions, logger);        for (Entry<AttributeKey<?>, Object> e: childAttrs) {          child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());      }        try {          // 將channel註冊到childGroup中的Selector上          childGroup.register(child).addListener(new ChannelFutureListener() {              @Override              public void operationComplete(ChannelFuture future) throws Exception {                  if (!future.isSuccess()) {                      forceClose(child, future.cause());                  }              }          });      } catch (Throwable t) {          forceClose(child, t);      }  }

READ事件

// NioByteUnsafe  public final void read() {      final ChannelConfig config = config();      final ChannelPipeline pipeline = pipeline();      final ByteBufAllocator allocator = config.getAllocator();      final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();      allocHandle.reset(config);        ByteBuf byteBuf = null;      boolean close = false;      try {          do {              byteBuf = allocHandle.allocate(allocator);              // 從channel中讀取數據，存放到byteBuf中              allocHandle.lastBytesRead(doReadBytes(byteBuf));                allocHandle.incMessagesRead(1);              readPending = false;                // 觸發fireChannelRead              pipeline.fireChannelRead(byteBuf);              byteBuf = null;          } while (allocHandle.continueReading());            // 觸發fireChannelReadComplete，如果在fireChannelReadComplete中執行了ChannelHandlerContext.flush，則響應結果返回給客戶端          allocHandle.readComplete();          // 觸發fireChannelReadComplete          pipeline.fireChannelReadComplete();            if (close) {              closeOnRead(pipeline);          }      } catch (Throwable t) {          if (!readPending && !config.isAutoRead()) {              removeReadOp();          }      }  }

寫事件

正常情況下一般是不會註冊寫事件的，如果Socket發送緩衝區中沒有空閑記憶體時，再寫入會導致阻塞，此時可以註冊寫事件，當有空閑記憶體（或者可用位元組數大於等於其低水位標記）時，再響應寫事件，並觸發對應回調。

if ((readyOps & SelectionKey.OP_WRITE) != 0) {      // 寫事件，從flush操作來看，雖然之前沒有向socket緩衝區寫數據，但是已經寫入到      // 了chnanel的outboundBuffer中，flush操作是將數據從outboundBuffer寫入到      // socket緩衝區      ch.unsafe().forceFlush();  }

CONNECT事件

該事件是client觸發的，由主動建立連接這一側觸發的。

if ((readyOps & SelectionKey.OP_CONNECT) != 0) {      // OP_CONNECT事件，client連接上客戶端時觸發的事件      int ops = k.interestOps();      ops &= ~SelectionKey.OP_CONNECT;      k.interestOps(ops);        // 觸發finishConnect事件，其中就包括fireChannelActive事件，如果有自定義的handler有channelActive方法，則會觸發      unsafe.finishConnect();  }

推薦閱讀

歡迎小夥伴關注【TopCoder】閱讀更多精彩好文。

Tags: