Netty原始碼分析之ChannelPipeline—出站事件的傳播
上篇文章中我們梳理了ChannelPipeline中入站事件的傳播，這篇文章中我們看下出站事件的傳播，也就是ChannelOutboundHandler介面的實現。
1、出站事件的傳播示例
我們對上篇文章中的示例程式碼進行改造，在ChannelPipeline中加入ChannelOutboundHandler出站實現
public class ServerApp {
public static void main(String[] args) {
EventLoopGroup boss = new NioEventLoopGroup();
EventLoopGroup work = new NioEventLoopGroup(2);
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(boss, work).channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
// p.addLast(new LoggingHandler(LogLevel.INFO));
// 向ChannelPipeline中新增自定義channelHandler
p.addLast(new OutHandlerA());
p.addLast(new ServerHandlerA());
p.addLast(new ServerHandlerB());
p.addLast(new ServerHandlerC());
p.addLast(new OutHandlerB());
p.addLast(new OutHandlerC());
}
});
bootstrap.bind(8050).sync();
} catch (Exception e) {
// TODO: handle exception
}
}
}
public class OutHandlerA extends ChannelOutboundHandlerAdapter {
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
System.err.println(this.getClass().getName()+msg);
ctx.writeAndFlush((ByteBuf)msg);
}
}
public class OutHandlerB extends ChannelOutboundHandlerAdapter {
@Override
public void write(ChannelHandlerContext ctx,Object msg,ChannelPromise promise) {
System.out.println(this.getClass().getName()+msg);
ctx.write((ByteBuf)msg);
}
}
public class OutHandlerC extends ChannelOutboundHandlerAdapter {
@Override
public void write(ChannelHandlerContext ctx,Object msg,ChannelPromise promise) {
System.out.println(this.getClass().getName()+"--"+msg);
ctx.write((ByteBuf)msg);
}
}然後我們在ServerHandlerA的channelRead方法中執行ctx的write方法，模擬訊息出站事件的發生。
public class ServerHandlerA extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object object) {
ByteBuf byteBuf = PooledByteBufAllocator.DEFAULT.buffer();
byteBuf.writeByte(1);
byteBuf.writeByte(2);
ctx.channel().write(byteBuf);
//ctx.write(byteBuf);
}
｝上面channelRead方法中write方法的呼叫有兩種方式 ctx.channel().write 與 ctx.write，這兩種方式有何區別呢，我們首先看下這兩種方式的執行結果
ctx.channel().write
io.netty.example.echo.my.OutHandlerC--PooledUnsafeDirectByteBuf(ridx: 0, widx: 2, cap: 256) io.netty.example.echo.my.OutHandlerB--PooledUnsafeDirectByteBuf(ridx: 0, widx: 2, cap: 256) io.netty.example.echo.my.OutHandlerA--PooledUnsafeDirectByteBuf(ridx: 0, widx: 2, cap: 256)
ctx.write
io.netty.example.echo.my.OutHandlerA--PooledUnsafeDirectByteBuf(ridx: 0, widx: 2, cap: 256)
可以看到當呼叫ctx.channel().write時，訊息在管道中傳播的順序是從尾部一直傳遞到最上層的OutboundHandler；而 ctx.write會從所在的 handler 向前找 OutboundHandler。
那麼這兩種方式區別是否就如結果所示呢，下面我們就開始對這兩種方法的內部實現進行分析
2、出站事件傳播的分析
ctx.channel().write與 ctx.write 分別用的是AbstractChannel與AbstractChannelHandlerContext的write方法
AbstractChannel 的 write方法
@Override
public ChannelFuture write(Object msg) {
return pipeline.write(msg);
}
AbstractChannelHandlerContext 的 write方法
@Override
public ChannelFuture write(Object msg) {
return write(msg, newPromise());
}
上面程式碼中AbstractChannel的 wirte方法最終呼叫的是pipeline的write方法，我們進入pipeline內部檢視，可以看到pipeline的write方法預設從尾部AbstractChannelHandlerContext節點開始呼叫。
@Override
public final ChannelFuture write(Object msg) {
return tail.write(msg);
}
繼續向下跟蹤最終它們呼叫的都是AbstractChannelHandlerContext 的 write方法，下面我們看下方法內部的具體實現。
private void write(Object msg, boolean flush, ChannelPromise promise) {
ObjectUtil.checkNotNull(msg, "msg");
try {
if (isNotValidPromise(promise, true)) {//檢查ChannelPromise是否有效
ReferenceCountUtil.release(msg);
// cancelled
return;
}
} catch (RuntimeException e) {
ReferenceCountUtil.release(msg);
throw e;
}
//尋找上一個AbstractChannelHandlerContext節點
AbstractChannelHandlerContext next = findContextOutbound();
final Object m = pipeline.touch(msg, next);
EventExecutor executor = next.executor();
if (executor.inEventLoop()) {//與當前執行緒是否一致
if (flush) {//確定是否要把資料沖刷到遠端節點
next.invokeWriteAndFlush(m, promise);
} else {
next.invokeWrite(m, promise);
}
} else { //如果不一致的封裝成writeTask任務執行緒
final AbstractWriteTask task;
if (flush) {
task = WriteAndFlushTask.newInstance(next, m, promise);
} else {
task = WriteTask.newInstance(next, m, promise);
}
//把該執行緒任務交給對應的EventExecutor執行
if (!safeExecute(executor, task, promise, m)) {
// We failed to submit the AbstractWriteTask. We need to cancel it so we decrement the pending bytes
// and put it back in the Recycler for re-use later.
//
// See https://github.com/netty/netty/issues/8343.
task.cancel();
}
}
}
主要關注下findContextOutbound()，這個方法的作用就是獲取當前AbstractChannelHandlerContext節點的上一個節點prev
private AbstractChannelHandlerContext findContextOutbound() {
AbstractChannelHandlerContext ctx = this;
do {
ctx = ctx.prev;//獲取當前節點的上一個節點
} while (!ctx.outbound);//判斷是不是出站節點
return ctx;
}
最終通過next.invokeWrite(m, promise)回撥方法，呼叫下一個節點中封裝的ChannelOutboundHandler的write方法，從而實現write方法事件的傳遞
private void invokeWrite(Object msg, ChannelPromise promise) {
if (invokeHandler()) {//判斷當前ChannelOutboundHandler是否已經被新增到pipeline中（handlerAdded事件觸發）
invokeWrite0(msg, promise);
} else {
write(msg, promise);
}
}
private boolean invokeHandler() {
// Store in local variable to reduce volatile reads.
int handlerState = this.handlerState;
return handlerState == ADD_COMPLETE || (!ordered && handlerState == ADD_PENDING);
}
private void invokeWrite0(Object msg, ChannelPromise promise) {
try {
((ChannelOutboundHandler) handler()).write(this, msg, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
}
到這裡整個出站事件的傳播流程已經基本清晰了，wirte方法本身就是一個尋找並回調下一個節點中wirte方法的過程。
3、write與writeAndFlush
在上面程式碼中可以看到這兩個方法主要在於是否會在執行write方法後，是否會執行flush方法。
private void invokeWriteAndFlush(Object msg, ChannelPromise promise) {
if (invokeHandler()) { //是否呼叫回撥方法
//呼叫write與flush回撥方法，最終呼叫自定義hander的對應實現
invokeWrite0(msg, promise);
invokeFlush0();
} else {
writeAndFlush(msg, promise);
}
}
這裡需要注意的是invokeFlush0()在invokeWrite0後執行，也就是必須等到訊息出站事件傳遞完畢後，才會呼叫flush把資料沖刷到遠端節點。簡單理解就是你無論是在OutHandlerA、OutHandlerB還是OutHandlerC中呼叫writeAndFlush，最後都是要在write事件傳遞完畢才會flush資料的。
同時我們需要注意到當write與flush事件從OutHandlerA再往上傳遞時，OutHandlerA的的上一個節點就是Pipeline的頭節點HeadContext，我們看下HeadContext的write與flush方法實現；
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
unsafe.write(msg, promise);
}
@Override
public void flush(ChannelHandlerContext ctx) {
unsafe.flush();
}
到這裡我們可以看出，訊息的真正入隊與傳送最終是通過HeadContext的write與flush方法實現。
通過以上的分析我們可以看到Pipeline出站事件的傳播流程，同時我們需要注意ctx.write與ctx.channel().write的區別以及訊息的傳送最終是通頭部節點呼叫unsafe的write與flush方法實現的，其中如有不足與不正確的地方還望指出與海涵。
關注微信公眾號，檢視更多技術文章。