Categories
程式開發

Netty源碼解析– ChannelPipeline機制與讀寫過程


本文繼續閱讀Netty源碼,解析ChannelPipeline事件傳播原理,以及Netty讀寫過程。

源碼分析基於Netty 4.1

ChannelPipeline

Netty中的ChannelPipeline可以理解為攔截器鏈,維護了一個ChannelHandler鍊錶,ChannelHandler即具體攔截器,可以在讀寫過程中,對數據進行處理。

ChannelHandler也可以分為兩類。

ChannelInboundHandler,監控Channel狀態變化,如channelActive,channelRegistered,通常通過重寫ChannelOutboundHandler#channelRead方法處理讀取到的數據,如HttpObjectDecoder將讀取到的數據解析為(netty)HttpRequest。

ChannelOutboundHandler,攔截IO事件,如bind,connect,read,write,通常通過重寫ChannelInboundHandler#write方法處理將寫入Channel的數據。如HttpResponseEncoder,將待寫入的數據轉換為Http格式。

ChannelPipeline的默認實現類為DefaultChannelPipeline,它在ChannelHandler鍊錶首尾維護了兩個特殊的ChannelHandler — HeadContext,TailContext。

HeadContext負責將IO事件轉發給對應的UnSafe處理,例如前面文章中說到的register,bind,read等操作。

TailContext主要是一些兜底處理,如channelRead方法釋放ByteBuf的引用等。

事件傳播

ChannelOutboundInvoker負責觸發ChannelOutboundHandler的方法,他們方法名相同,只是ChannelOutboundInvoker方法中少了ChannelHandlerContext參數。

同樣,ChannelInboundInvoker負責觸發ChannelInboundHandler的方法,但ChannelInboundInvoker的方法名多了fire,如ChannelInboundInvoker#fireChannelRead方法,觸發ChannelInboundHandler#channelRead。

ChannelPipeline和*ChannelHandlerContext*都繼承了這兩個接口。

但他們作用不同,ChannelPipeline是攔截器鏈,實際請求委託給ChannelHandlerContext處理。

ChannelHandlerContext接口(即ChannelHandler上下文)維護了鍊錶的上下節點,它作為ChannelHandler方法參數, 負責與ChannelPipeline及其他ChannelHandler互動。通過它可以動態修改Channel的屬性,給EventLoop提交任務,也可以向下一個(上一個)ChannelHandler傳播事件。

例如,在ChannelInboundHandler#channelRead處理完數據後,可以通過ChannelHandlerContext#write將數據寫到Channel。

ChannelInboundHandler#handler方法返回真正的ChannelHandler,並使用該ChannelHandler執行實際操作。

通過DefaultChannelPipeline#addFirst等方法添加ChannelHandler時,Netty會為ChannelHandler構造一個DefaultChannelHandlerContext,handler方法返回對應的ChannelHandler。

HeadContext,TailContext也實現了AbstractChannelHandlerContext,handler方法返回自身this。

我們也可以通過ChannelHandlerContext給EventLoop提交異步任務

ctx.channel().eventLoop().execute(new Runnable() {
public void run() {
...
}
});

對於阻塞時間較長的操作,使用異步任務完成是不錯的選擇。

下面以DefaultChannelPipeline#fireChannelRead為例,看一下他們的事件傳播過程。

DefaultChannelPipeline

public final ChannelPipeline fireChannelRead(Object msg) {
AbstractChannelHandlerContext.invokeChannelRead(head, msg);
return this;
}

使用HeadContext作為開始節點,調用AbstractChannelHandlerContext#invokeChannelRead方法開始調用攔截器鍊錶。

AbstractChannelHandlerContext

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
EventExecutor executor = next.executor();
if (executor.inEventLoop()) {
next.invokeChannelRead(m);
} else {
...
}
}

private void invokeChannelRead(Object msg) {
if (invokeHandler()) {
try {
// #1
((ChannelInboundHandler) handler()).channelRead(this, msg);
} catch (Throwable t) {
notifyHandlerException
}
} else {
fireChannelRead(msg);
}
}

#1

handler方法獲取AbstractChannelHandlerContext真正的Handler,再觸發其ChannelPipeline#channelRead方法

由於invokeChannelRead方法在HeadContext中執行,handler()這裡返回HeadContext,這時會觸發HeadContext#channelRead

HeadContext#channelRead

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ctx.fireChannelRead(msg);
}

HeadContext方法調用ctx.fireChannelRead(msg),就是向下一個ChannelInboundHandler傳播事件。

AbstractChannelHandlerContext#fireChannelRead

public ChannelHandlerContext fireChannelRead(final Object msg) {
invokeChannelRead(findContextInbound(MASK_CHANNEL_READ), msg);
return this;
}

AbstractChannelHandlerContext#fireChannelRead(final Object msg)方法主要負責找到下一個ChannelInboundHandler,並觸發其channelRead方法。

從DefaultChannelPipeline#fireChannelRead方法可以看到一個完整的調用鏈路:

#1 DefaultChannelPipeline通過HeadContext開始調用

#2 ChannelInboundHandler處理完當前邏輯後,調用ctx.fireChannelRead(msg)向後傳播事件

#3 AbstractChannelHandlerContext找到下一個ChannelInboundHandler,並觸發其channelRead,從而保證攔截器鏈繼續執行。

注意:對於ChannelOutboundHandler中的方法,DefaultChannelPipeline從TailContext開始調用,並向前傳播事件,與ChannelInboundHandler方向相反。

大家在閱讀Netty源碼時,對於DefaultChannelPipeline的方法,要注意該方法底層調用是ChannelInboundHandler還是ChannelOutboundHandler的方法,以及他們的傳播方向。

如果我們定義一個Http迴聲程序,示意代碼如下

new ServerBootstrap().group(parentGroup, childGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer() {
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new HttpRequestDecoder());
p.addLast(new HttpResponseEncoder());
p.addLast(new LoggingHandler(LogLevel.INFO));
p.addLast(new HttpEchoHandler());
}
});

其中HttpEchoHandler實現了ChannelInboundHandler,並在channelRead方法中調用ChannelHandlerContext#write方法回傳數據。

那麼,數據流轉如下所示

Socket.read() -> head#channelRead -> HttpRequestDecoder#channelRead -> LoggingHandler#channelRead -> HttpEchoHandler#channelRead
|
|/
Socket.write() <- head#write <- HttpResponseEncoder#write <- LoggingHandler#write <- ChannelHandlerContext#write

ChannelHandlerContext#write和DefaultChannelPipeline#write不同,前者從當前節點向前找到一個ChannelOutboundHandler開始調用,而後者則是從tail開始調用。

前面文章《事件循環機制實現原理》中說過,NioEventLoop#processSelectedKey中,通過NioUnsafe#read方法處理accept和read事件。下面來看一些read事件的處理。

NioByteUnsafe#read

public final void read() {
final ChannelConfig config = config();
if (shouldBreakReadReady(config)) {
clearReadPending();
return;
}
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 {
// #1
byteBuf = allocHandle.allocate(allocator);
// #2
allocHandle.lastBytesRead(doReadBytes(byteBuf));
// #3
if (allocHandle.lastBytesRead() <= 0) { byteBuf.release(); byteBuf = null; close = allocHandle.lastBytesRead() < 0; if (close) { readPending = false; } break; } allocHandle.incMessagesRead(1); readPending = false; // #4 pipeline.fireChannelRead(byteBuf); byteBuf = null; // #5 } while (allocHandle.continueReading()); // #6 allocHandle.readComplete(); // #7 pipeline.fireChannelReadComplete(); if (close) { // #8 closeOnRead(pipeline); } } catch (Throwable t) { handleReadException(pipeline, byteBuf, t, close, allocHandle); } finally { ... } }

#1 分配內存給ByteBuf

#2 讀取Socket數據到ByteBuf,這裡默認會嘗試讀取1024字節的數據。

#3 如果lastBytesRead方法返回-1,表示Channel已關閉,這時釋放當前ByteBuf引用,準備關閉Channel

#4 使用讀取到的數據,觸發ChannelPipeline#fireChannelRead,通常我們在這里處理數據。

#5 判斷是否需要繼續讀取數據。

默認條件是,如果讀取到的數據大小等於嘗試讀取數據大小1024字節,則繼續讀取。

#6 預留方法,提供給RecvByteBufAllocator做一些擴展操作

#7 觸發ChannelPipeline#fireChannelReadComplete,例如將前面多次讀取到的數據轉換為一個對象。

#8 關閉Channel

注意,ChannelPipeline#fireChannelRead如果不再繼續傳播channelRead事件,就不會執行到TailContext#channelRead方法,這是我們需要自行釋放對應的ByteBuf。

可以通過繼承SimpleChannelInboundHandler類實現,SimpleChannelInboundHandler#channelRead保證最終釋放ByteBuf。

我們需要調用ChannelHandlerContext#write方法觸發write操作。

ChannelHandlerContext#write-> HeadContext#write-> AbstractUnsafe#write

public final void write(Object msg, ChannelPromise promise) {
assertEventLoop();
// #1
ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
...

int size;
try {
// #2
msg = filterOutboundMessage(msg);
// #3
size = pipeline.estimatorHandle().size(msg);
if (size < 0) { size = 0; } } catch (Throwable t) { safeSetFailure(promise, t); ReferenceCountUtil.release(msg); return; } // #4 outboundBuffer.addMessage(msg, size, promise); }

#1 獲取AbstractUnsafe中維護的ChannelOutboundBuffer,該類負責緩存write的數據,等到flush再實際寫數據。

#2 AbstractChannel提供給子類的擴展方法,可以做一些ByteBuf檢查,轉化等操作。

#3 檢查待寫入數據量

#4 將數據添加到ChannelOutboundBuffer緩存中。

可以看到,write並沒有真正的寫數據,而是將數據放到了一個緩衝對象ChannelOutboundBuffer。

ChannelOutboundBuffer中的數據要​​等到ChannelHandlerContext#flush時再寫出。

ByteBuf是Netty中負責與Channel交互的內存緩衝區,而ByteBufAllocator,RecvByteBufAllocator主要負責分配內存給ByteBuf,後面有文章解析它們。

ChannelOutboundBuffer主要是緩存write數據,等到flush時再一併寫入Channel。後面有文章解析它。

如果您覺得本文不錯,歡迎關注我的微信公眾號,您的關注是我堅持的動力!

Netty源碼解析-- ChannelPipeline機制與讀寫過程 1