Netty源码分析之ChannelPipeline—入站事件的传播
bigfan 人气:0之前的文章中我们说过ChannelPipeline作为Netty中的数据管道，负责传递Channel中消息的事件传播，事件的传播分为入站和出站两个方向，分别通知ChannelInboundHandler与ChannelOutboundHandler来触发对应事件。这篇文章我们先对Netty中入站事件的传播，也就是ChannelInboundHandler进行下分析：
1、入站事件传播示例
我们通过一个简单的例子看下ChannelPipeline中入站事件channelRead的传播
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 ServerHandlerA());
p.addLast(new ServerHandlerB());
p.addLast(new ServerHandlerC());
}
});
bootstrap.bind(8050).sync();
} catch (Exception e) {
// TODO: handle exception
}
}
}
public class ServerHandlerA extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object object) {
System.out.println(this.getClass().getName() + "--"+object.toString());
ctx.fireChannelRead(object);
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
ctx.channel().pipeline().fireChannelRead("hello word");
}
}
public class ServerHandlerB extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object object) {
System.out.println(this.getClass().getName() + "--"+object.toString());
ctx.fireChannelRead(object);
}
}
public class ServerHandlerC extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object object) {
System.out.println(this.getClass().getName() + "--"+object.toString());
ctx.fireChannelRead(object);
}
}
客户端连接服务后可看到输出结果
io.netty.example.echo.my.ServerHandlerA--hello word io.netty.example.echo.my.ServerHandlerB--hello word io.netty.example.echo.my.ServerHandlerC--hello word
通过输出结果我们可以看到，消息会根据向ChannelPipeline中添加自定义channelHandler的顺序传递，并通过实现channelRead接口处理消息接收事件的。在例子中channelRead事件的传递是通过ctx.fireChannelRead(object)方法实现，接下来我们就从这里入手看下ChannelPipeline事件传递的具体实现。
2、channelRead事件的传播
首先这里需要注意的是我们例子中第一个节点的传递与实际应用中入站数据的传递是通过ChannelPipeline的fireChannelRead方法实现的，因为在实际的应用中，入站事件的传递是由NioUnsafe的read接口实现发起的，需要保证消息是从head结点开始传递的，例子中是为了模拟这一过程。
ctx.channel().pipeline().fireChannelRead("hello word");
@Override
public final ChannelPipeline fireChannelRead(Object msg) {
AbstractChannelHandlerContext.invokeChannelRead(head, msg);//默认传入head节点
return this;
}
进入invokeChannelRead方法内部看下具体实现；
static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
//ObjectUtil.checkNotNull 判断传入的消息数据是否为空
//next.pipeline.touch 对消息类型进行判断
final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
EventExecutor executor = next.executor();//获取ChannelHandlerContext对应的线程
if (executor.inEventLoop()) {//是否为当前线程
next.invokeChannelRead(m);//调用ChannelHandlerContext中invokeChannelRead的回调方法
} else {
executor.execute(new Runnable() {//如果线程不是当前线程
@Override
public void run() {
next.invokeChannelRead(m);
}
});
}
}
其中invokeChannelRead方法会获取该ChannelHandlerContext所封装的handler实现；
private void invokeChannelRead(Object msg) {
if (invokeHandler()) {
try {
//获取封装的ChannelInboundHandler实现，并调用我们实现的channelRead方法，
((ChannelInboundHandler) handler()).channelRead(this, msg);
} catch (Throwable t) {
notifyHandlerException(t);
}
} else {
fireChannelRead(msg);
}
}
前面我们知道首先传入的ChannelPipeline中ChannelHandlerContext链表的head头部节点HeadContext，看下其channelRead的方法实现；
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ctx.fireChannelRead(msg);
}
调用当前ChannelHandlerContext的fireChannelRead方法，进入ctx.fireChannelRead(object)方法内部看下具体的源码实现；
@Override
public ChannelHandlerContext fireChannelRead(final Object msg) {
//开始消息传递，findContextInbound方法按顺序获取当前ChannelHandlerContext的next节点
invokeChannelRead(findContextInbound(), msg);
return this;
}
findContextInbound方法获取的是HeadContext的下一个节点，也就是我们例子中向ChannelPipeline中添加自定义ServerHandlerA；
到这里其实就可以看出Pipeline中channelRead事件的传播主要就是通过ctx.fireChannelRead(msg)，获取当前ChannelHandlerContext下一个节点中封装的ChannelInboundHandler来实现的，最后一步一步传递到Tail尾部节点。
3、资源的释放及SimpleChannelInboundHandler
Netty中对象的生命周期由它们的引用计数管理的，为保证入站对象资源被释放，我们需要通过ReferenceCountUtil.release方法减少引用计数，确保对象的的最终计数器最后被置为0，从而被回收释放。我们看下Netty在入站事件中默认是如何减少引用计数的。
第一种方法，如果我们跟上面示例一样，在实现的每一个ChannelInboundHandler中都调用了ctx.fireChannelRead(msg)，最后消息会被传递到Tail尾节点，我们看下Tail节点中的channelRead方法
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
onUnhandledInboundMessage(msg);
}
protected void onUnhandledInboundMessage(Object msg) {
try {
logger.debug(
"Discarded inbound message {} that reached at the tail of the pipeline. " +
"Please check your pipeline configuration.", msg);
} finally {
ReferenceCountUtil.release(msg);
}
}
Tail节点的channelRead方法最终会调用ReferenceCountUtil.release方法来减少引用计数的，所以如果你在处理入站消息的过程中没有增加引用并且通过ctx.fireChannelRead(msg)方法把消息传到了Tail节点，你就不需要自己显式调用ReferenceCountUtil.release方法了。
其次如果继承的是SimpleChannelInboundHandler，可以看到SimpleChannelInboundHandler的channelRead方法实现中也已经调用了ReferenceCountUtil.release方法来减少引用计数；
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
boolean release = true;
try {
if (acceptInboundMessage(msg)) {
@SuppressWarnings("unchecked")
I imsg = (I) msg;
channelRead0(ctx, imsg);
} else {
release = false;
ctx.fireChannelRead(msg);
}
} finally {
if (autoRelease && release) {
ReferenceCountUtil.release(msg);
}
}
}
所以关于入站消息的资源释放方式总结如下：
- 1、继承ChannelInboundHandlerAdapter ，在channelRead的方法实现中调用ctx.fireChannelRead(object)方法，把消息一直向下传递，直到传递到Tail尾部节点，由Tail节点执行 ReferenceCountUtil.release来减少计数器，保证资源释放；
- 2、继承SimpleChannelInboundHandler，SimpleChannelInboundHandler本身的ChannelRead方法中会执行 ReferenceCountUtil.release来减少引用；
- 3、如果以上两点都没有做到，那就需要手动调用ReferenceCountUtil.release来减少引用来释放资源；
到这里我们基本了解了ChannelPipeline中入站事件是如何传播与相应的的，以及Netty中入站消息的资源释放机制。其中如有不足与不正确的地方还望指出与海涵。
关注微信公众号，查看更多技术文章。
加载全部内容