Netty源码分析之ChannelPipeline—出站事件的传…
2020-03-16 16:05:42来源:博客园 阅读 ()
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方法实现的,其中如有不足与不正确的地方还望指出与海涵。
关注微信公众号,查看更多技术文章。
原文链接:https://www.cnblogs.com/dafanjoy/p/12433929.html
如有疑问请与原作者联系
标签:
版权申明:本站文章部分自网络,如有侵权,请联系:west999com@outlook.com
特别注意:本站所有转载文章言论不代表本站观点,本站所提供的摄影照片,插画,设计作品,如需使用,请与原作者联系,版权归原作者所有
下一篇:细数MQ那些不得不说的8大好处
- 你说研究过Spring里面的源码,循环依赖你会么? 2020-06-09
- 这可能是目前最透彻的Netty讲解了... 2020-06-08
- 通俗理解spring源码(六)—— 默认标签(import、alias、be 2020-06-07
- 数据结构:用实例分析ArrayList与LinkedList的读写性能 2020-06-04
- 学习源码的第八个月,我成了Spring的开源贡献者 2020-06-02
IDC资讯: 主机资讯 注册资讯 托管资讯 vps资讯 网站建设
网站运营: 建站经验 策划盈利 搜索优化 网站推广 免费资源
网络编程: Asp.Net编程 Asp编程 Php编程 Xml编程 Access Mssql Mysql 其它
服务器技术: Web服务器 Ftp服务器 Mail服务器 Dns服务器 安全防护
软件技巧: 其它软件 Word Excel Powerpoint Ghost Vista QQ空间 QQ FlashGet 迅雷
网页制作: FrontPages Dreamweaver Javascript css photoshop fireworks Flash