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(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) 向后传播事件

#3 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。后面有文章解析它。

如果您觉得本文不错，欢迎关注我的微信公众号，您的关注是我坚持的动力！