Netty源码解析 -- ChannelPipeline机制与读写过程

binecy 2020-11-07 21:19:19
源码 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。
ChannelPipelineChannelHandlerContext都继承了这两个接口。
但他们作用不同,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(t);
}
} 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)向后传播事件
#4 AbstractChannelHandlerContext找到下一个ChannelInboundHandler,并触发其channelRead,从而保证拦截器链继续执行。

注意:对于ChannelOutboundHandler中的方法,DefaultChannelPipeline从TailContext开始调用,并向前传播事件,与ChannelInboundHandler方向相反。
大家在阅读Netty源码时,对于DefaultChannelPipeline的方法,要注意该方法底层调用是ChannelInboundHandler还是ChannelOutboundHandler的方法,以及他们的传播方向。

如果我们定义一个Http回声程序,示意代码如下

new ServerBootstrap().group(parentGroup, childGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
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开始调用。

Read

前面文章《事件循环机制实现原理》中说过,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。

Write

我们需要调用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。后面有文章解析它。

如果您觉得本文不错,欢迎关注我的微信公众号,系列文章持续更新中。您的关注是我坚持的动力!

版权声明
本文为[binecy]所创,转载请带上原文链接,感谢
https://www.cnblogs.com/binecy/p/13942384.html

  1. 【计算机网络 12(1),尚学堂马士兵Java视频教程
  2. 【程序猿历程,史上最全的Java面试题集锦在这里
  3. 【程序猿历程(1),Javaweb视频教程百度云
  4. Notes on MySQL 45 lectures (1-7)
  5. [computer network 12 (1), Shang Xuetang Ma soldier java video tutorial
  6. The most complete collection of Java interview questions in history is here
  7. [process of program ape (1), JavaWeb video tutorial, baidu cloud
  8. Notes on MySQL 45 lectures (1-7)
  9. 精进 Spring Boot 03:Spring Boot 的配置文件和配置管理,以及用三种方式读取配置文件
  10. Refined spring boot 03: spring boot configuration files and configuration management, and reading configuration files in three ways
  11. 精进 Spring Boot 03:Spring Boot 的配置文件和配置管理,以及用三种方式读取配置文件
  12. Refined spring boot 03: spring boot configuration files and configuration management, and reading configuration files in three ways
  13. 【递归,Java传智播客笔记
  14. [recursion, Java intelligence podcast notes
  15. [adhere to painting for 386 days] the beginning of spring of 24 solar terms
  16. K8S系列第八篇(Service、EndPoints以及高可用kubeadm部署)
  17. K8s Series Part 8 (service, endpoints and high availability kubeadm deployment)
  18. 【重识 HTML (3),350道Java面试真题分享
  19. 【重识 HTML (2),Java并发编程必会的多线程你竟然还不会
  20. 【重识 HTML (1),二本Java小菜鸟4面字节跳动被秒成渣渣
  21. [re recognize HTML (3) and share 350 real Java interview questions
  22. [re recognize HTML (2). Multithreading is a must for Java Concurrent Programming. How dare you not
  23. [re recognize HTML (1), two Java rookies' 4-sided bytes beat and become slag in seconds
  24. 造轮子系列之RPC 1:如何从零开始开发RPC框架
  25. RPC 1: how to develop RPC framework from scratch
  26. 造轮子系列之RPC 1:如何从零开始开发RPC框架
  27. RPC 1: how to develop RPC framework from scratch
  28. 一次性捋清楚吧,对乱糟糟的,Spring事务扩展机制
  29. 一文彻底弄懂如何选择抽象类还是接口,连续四年百度Java岗必问面试题
  30. Redis常用命令
  31. 一双拖鞋引发的血案,狂神说Java系列笔记
  32. 一、mysql基础安装
  33. 一位程序员的独白:尽管我一生坎坷,Java框架面试基础
  34. Clear it all at once. For the messy, spring transaction extension mechanism
  35. A thorough understanding of how to choose abstract classes or interfaces, baidu Java post must ask interview questions for four consecutive years
  36. Redis common commands
  37. A pair of slippers triggered the murder, crazy God said java series notes
  38. 1、 MySQL basic installation
  39. Monologue of a programmer: despite my ups and downs in my life, Java framework is the foundation of interview
  40. 【大厂面试】三面三问Spring循环依赖,请一定要把这篇看完(建议收藏)
  41. 一线互联网企业中,springboot入门项目
  42. 一篇文带你入门SSM框架Spring开发,帮你快速拿Offer
  43. 【面试资料】Java全集、微服务、大数据、数据结构与算法、机器学习知识最全总结,283页pdf
  44. 【leetcode刷题】24.数组中重复的数字——Java版
  45. 【leetcode刷题】23.对称二叉树——Java版
  46. 【leetcode刷题】22.二叉树的中序遍历——Java版
  47. 【leetcode刷题】21.三数之和——Java版
  48. 【leetcode刷题】20.最长回文子串——Java版
  49. 【leetcode刷题】19.回文链表——Java版
  50. 【leetcode刷题】18.反转链表——Java版
  51. 【leetcode刷题】17.相交链表——Java&python版
  52. 【leetcode刷题】16.环形链表——Java版
  53. 【leetcode刷题】15.汉明距离——Java版
  54. 【leetcode刷题】14.找到所有数组中消失的数字——Java版
  55. 【leetcode刷题】13.比特位计数——Java版
  56. oracle控制用户权限命令
  57. 三年Java开发,继阿里,鲁班二期Java架构师
  58. Oracle必须要启动的服务
  59. 万字长文!深入剖析HashMap,Java基础笔试题大全带答案
  60. 一问Kafka就心慌?我却凭着这份,图灵学院vip课程百度云