本系列Netty源码解析文章基于 4.1.56.Final版本
我们接着上篇文章一文搞懂Netty发送数据全流程 | 你想知道的细节全在这里继续讲解 Netty 的 flush 全流程。
4. flush
从前面 Netty 对 write 事件的处理过程中，我们可以看到当用户调用 ctx.write(msg) 方法之后，Netty 只是将用户要发送的数据临时写到 channel 对应的待发送缓冲队列 ChannelOutboundBuffer 中，然而并不会将数据写入 Socket 中。
而当一次 read 事件完成之后，我们会调用 ctx.flush() 方法将 ChannelOutboundBuffer 中的待发送数据写入 Socket 中的发送缓冲区中，从而将数据发送出去。
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
//本次OP_READ事件处理完毕
ctx.flush();
}
}
4.1 flush事件的传播
pipeline结构.png
flush 事件和 write 事件一样都是 oubound 事件，所以它们的传播方向都是从后往前在 pipeline 中传播。
触发 flush 事件传播的同样也有两个方法：
-
channelHandlerContext.flush()：flush事件会从当前 channelHandler 开始在 pipeline 中向前传播直到 headContext。 -
channelHandlerContext.channel().flush()：flush 事件会从 pipeline 的尾结点 tailContext 处开始向前传播直到 headContext。
abstract class AbstractChannelHandlerContext implements ChannelHandlerContext, ResourceLeakHint {
@Override
public ChannelHandlerContext flush() {
//向前查找覆盖flush方法的Outbound类型的ChannelHandler
final AbstractChannelHandlerContext next = findContextOutbound(MASK_FLUSH);
//获取执行ChannelHandler的executor,在初始化pipeline的时候设置，默认为Reactor线程
EventExecutor executor = next.executor();
if (executor.inEventLoop()) {
next.invokeFlush();
} else {
Tasks tasks = next.invokeTasks;
if (tasks == null) {
next.invokeTasks = tasks = new Tasks(next);
}
safeExecute(executor, tasks.invokeFlushTask, channel().voidPromise(), null, false);
}
return this;
}
}
这里的逻辑和 write 事件传播的逻辑基本一样，也是首先通过findContextOutbound(MASK_FLUSH) 方法从当前 ChannelHandler 开始从 pipeline 中向前查找出第一个 ChannelOutboundHandler 类型的并且实现 flush 事件回调方法的 ChannelHandler 。注意这里传入的执行资格掩码为 MASK_FLUSH。
执行ChannelHandler中事件回调方法的线程必须是通过pipeline#addLast(EventExecutorGroup group, ChannelHandler... handlers)为 ChannelHandler 指定的 executor。如果不指定，默认的 executor 为 channel 绑定的 reactor 线程。
如果当前线程不是 ChannelHandler 指定的 executor，则需要将 invokeFlush() 方法的调用封装成 Task 交给指定的 executor 执行。
4.1.1 触发nextChannelHandler的flush方法回调
private void invokeFlush() {
if (invokeHandler()) {
invokeFlush0();
} else {
//如果该ChannelHandler并没有加入到pipeline中则继续向前传递flush事件
flush();
}
}
这里和 write 事件的相关处理一样，首先也是需要调用 invokeHandler() 方法来判断这个 nextChannelHandler 是否在 pipeline 中被正确的初始化。
如果 nextChannelHandler 中的 handlerAdded 方法并没有被回调过，那么这里就只能跳过 nextChannelHandler，并调用 ChannelHandlerContext#flush 方法继续向前传播flush事件。
如果 nextChannelHandler 中的 handlerAdded 方法已经被回调过，说明 nextChannelHandler 在 pipeline 中已经被正确的初始化好，则直接调用nextChannelHandler 的 flush 事件回调方法。
private void invokeFlush0() {
try {
((ChannelOutboundHandler) handler()).flush(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
}
这里有一点和 write 事件处理不同的是，当调用 nextChannelHandler 的 flush 回调出现异常的时候，会触发 nextChannelHandler 的 exceptionCaught 回调。
private void invokeExceptionCaught(final Throwable cause) {
if (invokeHandler()) {
try {
handler().exceptionCaught(this, cause);
} catch (Throwable error) {
if (logger.isDebugEnabled()) {
logger.debug(....相关日志打印......);
} else if (logger.isWarnEnabled()) {
logger.warn(...相关日志打印......));
}
}
} else {
fireExceptionCaught(cause);
}
}
而其他 outbound 类事件比如 write 事件在传播的过程中发生异常，只是回调通知相关的 ChannelFuture。并不会触发 exceptionCaught 事件的传播。
4.2 flush事件的处理
客户端channel pipeline结构.png
最终flush事件会在pipeline中一直向前传播至HeadContext中，并在 HeadContext 里调用 channel 的 unsafe 类完成 flush 事件的最终处理逻辑。
final class HeadContext extends AbstractChannelHandlerContext {
@Override
public void flush(ChannelHandlerContext ctx) {
unsafe.flush();
}
}
下面就真正到了 Netty 处理 flush 事件的地方。
protected abstract class AbstractUnsafe implements Unsafe {
@Override
public final void flush() {
assertEventLoop();
ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
//channel以关闭
if (outboundBuffer == null) {
return;
}
//将flushedEntry指针指向ChannelOutboundBuffer头结点，此时变为即将要flush进Socket的数据队列
outboundBuffer.addFlush();
//将待写数据写进Socket
flush0();
}
}
4.2.1 ChannelOutboundBuffer#addFlush
ChannelOutboundBuffer结构.png
这里就到了真正要发送数据的时候了，在 addFlush 方法中会将 flushedEntry 指针指向 unflushedEntry 指针表示的第一个未被 flush 的 Entry 节点。并将 unflushedEntry 指针置为空，准备开始 flush 发送数据流程。
此时 ChannelOutboundBuffer 由待发送数据的缓冲队列变为了即将要 flush 进 Socket 的数据队列
这样在 flushedEntry 与 tailEntry 之间的 Entry 节点即为本次 flush 操作需要发送的数据范围。
public void addFlush() {
Entry entry = unflushedEntry;
if (entry != null) {
if (flushedEntry == null) {
flushedEntry = entry;
}
do {
flushed ++;
//如果当前entry对应的write操作被用户取消，则释放msg，并降低channelOutboundBuffer水位线
if (!entry.promise.setUncancellable()) {
int pending = entry.cancel();
decrementPendingOutboundBytes(pending, false, true);
}
entry = entry.next;
} while (entry != null);
// All flushed so reset unflushedEntry
unflushedEntry = null;
}
}
在 flush 发送数据流程开始时，数据的发送流程就不能被取消了，在这之前我们都是可以通过 ChannelPromise 取消数据发送流程的。
所以这里需要对 ChannelOutboundBuffer 中所有 Entry 节点包裹的 ChannelPromise 设置为不可取消状态。
public interface Promise<V> extends Future<V> {
/**
* 设置当前future为不可取消状态
*
* 返回true的情况：
* 1：成功的将future设置为uncancellable
* 2：当future已经成功完成
*
* 返回false的情况：
* 1：future已经被取消，则不能在设置 uncancellable 状态
*
*/
boolean setUncancellable();
}
如果这里的 setUncancellable() 方法返回 false 则说明在这之前用户已经将 ChannelPromise 取消掉了，接下来就需要调用 entry.cancel() 方法来释放为待发送数据 msg 分配的堆外内存。
static final class Entry {
//write操作是否被取消
boolean cancelled;
int cancel() {
if (!cancelled) {
cancelled = true;
int pSize = pendingSize;
// release message and replace with an empty buffer
ReferenceCountUtil.safeRelease(msg);
msg = Unpooled.EMPTY_BUFFER;
pendingSize = 0;
total = 0;
progress = 0;
bufs = null;
buf = null;
return pSize;
}
return 0;
}
}
当 Entry 对象被取消后，就需要减少 ChannelOutboundBuffer 的内存占用总量的水位线 totalPendingSize。
private static final AtomicLongFieldUpdater<ChannelOutboundBuffer> TOTAL_PENDING_SIZE_UPDATER =
AtomicLongFieldUpdater.newUpdater(ChannelOutboundBuffer.class, "totalPendingSize");
//水位线指针.ChannelOutboundBuffer中的待发送数据的内存占用总量 : 所有Entry对象本身所占用内存大小 + 所有待发送数据的大小
private volatile long totalPendingSize;
private void decrementPendingOutboundBytes(long size, boolean invokeLater, boolean notifyWritability) {
if (size == 0) {
return;
}
long newWriteBufferSize = TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, -size);
if (notifyWritability && newWriteBufferSize < channel.config().getWriteBufferLowWaterMark()) {
setWritable(invokeLater);
}
}
当更新后的水位线低于低水位线 DEFAULT_LOW_WATER_MARK = 32 * 1024 时，就将当前 channel 设置为可写状态。
private void setWritable(boolean invokeLater) {
for (;;) {
final int oldValue = unwritable;
final int newValue = oldValue & ~1;
if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
if (oldValue != 0 && newValue == 0) {
fireChannelWritabilityChanged(invokeLater);
}
break;
}
}
}
当 Channel 的状态是第一次从不可写状态变为可写状态时，Netty 会在 pipeline 中再次触发 ChannelWritabilityChanged 事件的传播。
响应channelWritabilityChanged事件.png
4.2.2 发送数据前的最后检查---flush0
flush0 方法这里主要做的事情就是检查当 channel 的状态是否正常，如果 channel 状态一切正常，则调用 doWrite 方法发送数据。
protected abstract class AbstractUnsafe implements Unsafe {
//是否正在进行flush操作
private boolean inFlush0;
protected void flush0() {
if (inFlush0) {
// Avoid re-entrance
return;
}
final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
//channel已经关闭或者outboundBuffer为空
if (outboundBuffer == null || outboundBuffer.isEmpty()) {
return;
}
inFlush0 = true;
if (!isActive()) {
try {
if (!outboundBuffer.isEmpty()) {
if (isOpen()) {
//当前channel处于disConnected状态 通知promise 写入失败 并触发channelWritabilityChanged事件
outboundBuffer.failFlushed(new NotYetConnectedException(), true);
} else {
//当前channel处于关闭状态 通知promise 写入失败 但不触发channelWritabilityChanged事件
outboundBuffer.failFlushed(newClosedChannelException(initialCloseCause, "flush0()"), false);
}
}
} finally {
inFlush0 = false;
}
return;
}
try {
//写入Socket
doWrite(outboundBuffer);
} catch (Throwable t) {
handleWriteError(t);
} finally {
inFlush0 = false;
}
}
}
-
outboundBuffer == null || outboundBuffer.isEmpty()：如果 channel 已经关闭了或者对应写缓冲区中没有任何数据，那么就停止发送流程，直接 return。 -
!isActive()：如果当前channel处于非活跃状态，则需要调用outboundBuffer#failFlushed通知 ChannelOutboundBuffer 中所有待发送操作对应的 channelPromise 向用户线程报告发送失败。并将待发送数据 Entry 对象从 ChannelOutboundBuffer 中删除，并释放待发送数据空间，回收 Entry 对象实例。
还记得我们在《Netty如何高效接收网络连接》一文中提到过的 NioSocketChannel 的 active 状态有哪些条件吗？？
@Override
public boolean isActive() {
SocketChannel ch = javaChannel();
return ch.isOpen() && ch.isConnected();
}
NioSocketChannel 处于 active 状态的条件必须是当前 NioSocketChannel 是 open 的同时处于 connected 状态。
-
!isActive() && isOpen()：说明当前 channel 处于 disConnected 状态，这时通知给用户 channelPromise 的异常类型为 NotYetConnectedException ,并释放所有待发送数据占用的堆外内存，如果此时内存占用量低于低水位线，则设置 channel 为可写状态，并触发 channelWritabilityChanged 事件。
当 channel 处于 disConnected 状态时，用户可以进行 write 操作但不能进行 flush 操作。
-
!isActive() && !isOpen()：说明当前 channel 处于关闭状态，这时通知给用户 channelPromise 的异常类型为 newClosedChannelException ，因为 channel 已经关闭，所以这里并不会触发 channelWritabilityChanged 事件。 -
当 channel 的这些异常状态校验通过之后，则调用 doWrite 方法将 ChannelOutboundBuffer 中的待发送数据写进底层 Socket 中。
4.2.2.1 ChannelOutboundBuffer#failFlushed
public final class ChannelOutboundBuffer {
private boolean inFail;
void failFlushed(Throwable cause, boolean notify) {
if (inFail) {
return;
}
try {
inFail = true;
for (;;) {
if (!remove0(cause, notify)) {
break;
}
}
} finally {
inFail = false;
}
}
}
该方法用于在 Netty 在发送数据的时候，如果发现当前 channel 处于非活跃状态，则将 ChannelOutboundBuffer 中 flushedEntry 与tailEntry 之间的 Entry 对象节点全部删除，并释放发送数据占用的内存空间，同时回收 Entry 对象实例。
4.2.2.2 ChannelOutboundBuffer#remove0
private boolean remove0(Throwable cause, boolean notifyWritability) {
Entry e = flushedEntry;
if (e == null) {
//清空当前reactor线程缓存的所有待发送数据
clearNioBuffers();
return false;
}
Object msg = e.msg;
ChannelPromise promise = e.promise;
int size = e.pendingSize;
//从channelOutboundBuffer中删除该Entry节点
removeEntry(e);
if (!e.cancelled) {
// only release message, fail and decrement if it was not canceled before.
//释放msg所占用的内存空间
ReferenceCountUtil.safeRelease(msg);
//编辑promise发送失败，并通知相应的Lisener
safeFail(promise, cause);
//由于msg得到释放，所以需要降低channelOutboundBuffer中的内存占用水位线，并根据notifyWritability决定是否触发ChannelWritabilityChanged事件
decrementPendingOutboundBytes(size, false, notifyWritability);
}
// recycle the entry
//回收Entry实例对象
e.recycle();
return true;
}
当一个 Entry 节点需要从 ChannelOutboundBuffer 中清除时，Netty 需要释放该 Entry 节点中包裹的发送数据 msg 所占用的内存空间。并标记对应的 promise 为失败同时通知对应的 listener ，由于 msg 得到释放，所以需要降低 channelOutboundBuffer 中的内存占用水位线，并根据 boolean notifyWritability 决定是否触发 ChannelWritabilityChanged 事件。最后需要将该 Entry 实例回收至 Recycler 对象池中。
5. 终于开始真正地发送数据了！
来到这里我们就真正进入到了 Netty 发送数据的核心处理逻辑，在《Netty如何高效接收网络数据》一文中，笔者详细介绍了 Netty 读取数据的核心流程，Netty 会在一个 read loop 中不断循环读取 Socket 中的数据直到数据读取完毕或者读取次数已满 16 次，当循环读取了 16 次还没有读取完毕时，Netty 就不能在继续读了，因为 Netty 要保证 Reactor 线程可以均匀的处理注册在它上边的所有 Channel 中的 IO 事件。剩下未读取的数据等到下一次 read loop 在开始读取。
除此之外，在每次 read loop 开始之前，Netty 都会分配一个初始化大小为 2048 的 DirectByteBuffer 来装载从 Socket 中读取到的数据，当整个 read loop 结束时，会根据本次读取数据的总量来判断是否为该 DirectByteBuffer 进行扩容或者缩容，目的是在下一次 read loop 的时候可以为其分配一个容量大小合适的 DirectByteBuffer 。
其实 Netty 对发送数据的处理和对读取数据的处理核心逻辑都是一样的，这里大家可以将这两篇文章结合对比着看。
但发送数据的细节会多一些，也会更复杂一些，由于这块逻辑整体稍微比较复杂，所以我们接下来还是分模块进行解析：
5.1 发送数据前的准备工作
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
//获取NioSocketChannel中封装的jdk nio底层socketChannel
SocketChannel ch = javaChannel();
//最大写入次数 默认为16 目的是为了保证SubReactor可以平均的处理注册其上的所有Channel
int writeSpinCount = config().getWriteSpinCount();
do {
if (in.isEmpty()) {
// 如果全部数据已经写完 则移除OP_WRITE事件并直接退出writeLoop
clearOpWrite();
return;
}
// SO_SNDBUF设置的发送缓冲区大小 * 2 作为 最大写入字节数 293976 = 146988 << 1
int maxBytesPerGatheringWrite = ((NioSocketChannelConfig) config).getMaxBytesPerGatheringWrite();
// 将ChannelOutboundBuffer中缓存的DirectBuffer转换成JDK NIO 的 ByteBuffer
ByteBuffer[] nioBuffers = in.nioBuffers(1024, maxBytesPerGatheringWrite);
// ChannelOutboundBuffer中总共的DirectBuffer数
int nioBufferCnt = in.nioBufferCount();
switch (nioBufferCnt) {
.........向底层jdk nio socketChannel发送数据.........
}
} while (writeSpinCount > 0);
............处理本轮write loop未写完的情况.......
}
这部分内容为 Netty 开始发送数据之前的准备工作：
5.1.1 获取write loop最大发送循环次数
从当前 NioSocketChannel 的配置类 NioSocketChannelConfig 中获取 write loop 最大循环写入次数，默认为 16。但也可以通过下面的方式进行自定义设置。
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.......
.childOption(ChannelOption.WRITE_SPIN_COUNT,自定义数值)
5.1.2 处理在一轮write loop中就发送完数据的情况
进入 write loop 之后首先需要判断当前 ChannelOutboundBuffer 中的数据是否已经写完了 in.isEmpty()) ，如果全部写完就需要清除当前 Channel 在 Reactor 上注册的 OP_WRITE 事件。
这里大家可能会有疑问，目前我们还没有注册 OP_WRITE 事件到 Reactor 上，为啥要清除呢？ 别着急，笔者会在后面为大家揭晓答案。
5.1.3 获取本次write loop 最大允许发送字节数
从 ChannelConfig 中获取本次 write loop 最大允许发送的字节数
maxBytesPerGatheringWrite 。初始值为 SO_SNDBUF大小 * 2 = 293976 = 146988 << 1，最小值为 2048。
private final class NioSocketChannelConfig extends DefaultSocketChannelConfig {
//293976 = 146988 << 1
//SO_SNDBUF设置的发送缓冲区大小 * 2 作为 最大写入字节数
//最小值为2048
private volatile int maxBytesPerGatheringWrite = Integer.MAX_VALUE;
private NioSocketChannelConfig(NioSocketChannel channel, Socket javaSocket) {
super(channel, javaSocket);
calculateMaxBytesPerGatheringWrite();
}
private void calculateMaxBytesPerGatheringWrite() {
// 293976 = 146988 << 1
// SO_SNDBUF设置的发送缓冲区大小 * 2 作为 最大写入字节数
int newSendBufferSize = getSendBufferSize() << 1;
if (newSendBufferSize > 0) {
setMaxBytesPerGatheringWrite(newSendBufferSize);
}
}
}
我们可以通过如下的方式自定义配置 Socket 发送缓冲区大小。
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.......
.childOption(ChannelOption.SO_SNDBUF,自定义数值)
5.1.4 将待发送数据转换成 JDK NIO ByteBuffer
由于最终 Netty 会调用 JDK NIO 的 SocketChannel 发送数据，所以这里需要首先将当前 Channel 中的写缓冲队列 ChannelOutboundBuffer 里存储的 DirectByteBuffer（ Netty 中的 ByteBuffer 实现）转换成 JDK NIO 的 ByteBuffer 类型。最终将转换后的待发送数据存储在 ByteBuffer[] nioBuffers 数组中。这里通过调用 ChannelOutboundBuffer#nioBuffers 方法完成以上 ByteBuffer 类型的转换。
-
maxBytesPerGatheringWrite：表示本次 write loop 中最多从 ChannelOutboundBuffer 中转换 maxBytesPerGatheringWrite 个字节出来。也就是本次 write loop 最多能发送多少字节。 -
1024: 本次 write loop 最多转换 1024 个 ByteBuffer（ JDK NIO 实现）。也就是说本次 write loop 最多批量发送多少个 ByteBuffer 。
通过 ChannelOutboundBuffer#nioBufferCount() 获取本次 write loop 总共需要发送的 ByteBuffer 数量 nioBufferCnt 。注意这里已经变成了 JDK NIO 实现的 ByteBuffer 了。
详细的 ByteBuffer 类型转换过程，笔者会在专门讲解 Buffer 设计的时候为大家全面细致地讲解，这里我们还是主要聚焦于发送数据流程的主线。
当做完这些发送前的准备工作之后，接下来 Netty 就开始向 JDK NIO SocketChannel 发送这些已经转换好的 JDK NIO ByteBuffer 了。
5.2 向JDK NIO SocketChannel发送数据
flush流程.png
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
SocketChannel ch = javaChannel();
int writeSpinCount = config().getWriteSpinCount();
do {
.........将待发送数据转换到JDK NIO ByteBuffer中.........
//本次write loop中需要发送的 JDK ByteBuffer个数
int nioBufferCnt = in.nioBufferCount();
switch (nioBufferCnt) {
case 0:
//这里主要是针对 网络传输文件数据 的处理 FileRegion
writeSpinCount -= doWrite0(in);
break;
case 1: {
.........处理单个NioByteBuffer发送的情况......
break;
}
default: {
.........批量处理多个NioByteBuffers发送的情况......
break;
}
}
} while (writeSpinCount > 0);
............处理本轮write loop未写完的情况.......
}
这里大家可能对 nioBufferCnt == 0 的情况比较有疑惑，明明之前已经校验过ChannelOutboundBuffer 不为空了，为什么这里从 ChannelOutboundBuffer 中获取到的 nioBuffer 个数依然为 0 呢？
在前边我们介绍 Netty 对 write 事件的处理过程时提过， ChannelOutboundBuffer 中只支持 ByteBuf 类型和 FileRegion 类型，其中 ByteBuf 类型用于装载普通的发送数据，而 FileRegion 类型用于通过零拷贝的方式网络传输文件。
而这里 ChannelOutboundBuffer 虽然不为空，但是装载的 NioByteBuffer 个数却为 0 说明 ChannelOutboundBuffer 中装载的是 FileRegion 类型，当前正在进行网络文件的传输。
case 0 的分支主要就是用于处理网络文件传输的情况。
5.2.1 零拷贝发送网络文件
protected final int doWrite0(ChannelOutboundBuffer in) throws Exception {
Object msg = in.current();
if (msg == null) {
return 0;
}
return doWriteInternal(in, in.current());
}
这里需要特别注意的是用于文件传输的方法 doWriteInternal 中的返回值，理解这些返回值的具体情况有助于我们理解后面 write loop 的逻辑走向。
private int doWriteInternal(ChannelOutboundBuffer in, Object msg) throws Exception {
if (msg instanceof ByteBuf) {
..............忽略............
} else if (msg instanceof FileRegion) {
FileRegion region = (FileRegion) msg;
//文件已经传输完毕
if (region.transferred() >= region.count()) {
in.remove();
return 0;
}
//零拷贝的方式传输文件
long localFlushedAmount = doWriteFileRegion(region);
if (localFlushedAmount > 0) {
in.progress(localFlushedAmount);
if (region.transferred() >= region.count()) {
in.remove();
}
return 1;
}
} else {
// Should not reach here.
throw new Error();
}
//走到这里表示 此时Socket已经写不进去了 退出writeLoop，注册OP_WRITE事件
return WRITE_STATUS_SNDBUF_FULL;
}
最终会在 doWriteFileRegion 方法中通过 FileChannel#transferTo 方法底层用到的系统调用为 sendFile 实现零拷贝网络文件的传输。
public class NioSocketChannel extends AbstractNioByteChannel implements io.netty.channel.socket.SocketChannel {
@Override
protected long doWriteFileRegion(FileRegion region) throws Exception {
final long position = region.transferred();
return region.transferTo(javaChannel(), position);
}
}
关于 Netty 中涉及到的零拷贝，笔者会有一篇专门的文章为大家讲解，本文的主题我们还是先聚焦于把发送流程的主线打通。
我们继续回到发送数据流程主线上来~~
case 0:
//这里主要是针对 网络传输文件数据 的处理 FileRegion
writeSpinCount -= doWrite0(in);
break;
-
region.transferred() >= region.count()：表示当前 FileRegion 中的文件数据已经传输完毕。那么在这种情况下本次 write loop 没有写入任何数据到 Socket ，所以返回 0 ，writeSpinCount - 0 意思就是本次 write loop 不算，继续循环。 -
localFlushedAmount > 0：表示本 write loop 中写入了一些数据到 Socket 中，会有返回 1，writeSpinCount - 1 减少一次 write loop 次数。 -
localFlushedAmount <= 0：表示当前 Socket 发送缓冲区已满，无法写入数据，那么就返回WRITE_STATUS_SNDBUF_FULL = Integer.MAX_VALUE。
writeSpinCount - Integer.MAX_VALUE必然是负数，直接退出循环，向 Reactor 注册 OP_WRITE 事件并退出 flush 流程。等 Socket 发送缓冲区可写了，Reactor 会通知 channel 继续发送文件数据。记住这里，我们后面还会提到。
5.2.2 发送普通数据
剩下两个 case 1 和 default 分支主要就是处理 ByteBuffer 装载的普通数据发送逻辑。
其中 case 1 表示当前 Channel 的 ChannelOutboundBuffer 中只包含了一个 NioByteBuffer 的情况。
default 表示当前 Channel 的 ChannelOutboundBuffer 中包含了多个 NioByteBuffers 的情况。
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
SocketChannel ch = javaChannel();
int writeSpinCount = config().getWriteSpinCount();
do {
.........将待发送数据转换到JDK NIO ByteBuffer中.........
//本次write loop中需要发送的 JDK ByteBuffer个数
int nioBufferCnt = in.nioBufferCount();
switch (nioBufferCnt) {
case 0:
..........处理网络文件传输.........
case 1: {
ByteBuffer buffer = nioBuffers[0];
int attemptedBytes = buffer.remaining();
final int localWrittenBytes = ch.write(buffer);
if (localWrittenBytes <= 0) {
//如果当前Socket发送缓冲区满了写不进去了，则注册OP_WRITE事件，等待Socket发送缓冲区可写时 在写
// SubReactor在处理OP_WRITE事件时，直接调用flush方法
incompleteWrite(true);
return;
}
//根据当前实际写入情况调整 maxBytesPerGatheringWrite数值
adjustMaxBytesPerGatheringWrite(attemptedBytes, localWrittenBytes, maxBytesPerGatheringWrite);
//如果ChannelOutboundBuffer中的某个Entry被全部写入 则删除该Entry
// 如果Entry被写入了一部分 还有一部分未写入 则更新Entry中的readIndex 等待下次writeLoop继续写入
in.removeBytes(localWrittenBytes);
--writeSpinCount;
break;
}
default: {
// ChannelOutboundBuffer中总共待写入数据的字节数
long attemptedBytes = in.nioBufferSize();
//批量写入
final long localWrittenBytes = ch.write(nioBuffers, 0, nioBufferCnt);
if (localWrittenBytes <= 0) {
incompleteWrite(true);
return;
}
//根据实际写入情况调整一次写入数据大小的最大值
// maxBytesPerGatheringWrite决定每次可以从channelOutboundBuffer中获取多少发送数据
adjustMaxBytesPerGatheringWrite((int) attemptedBytes, (int) localWrittenBytes,
maxBytesPerGatheringWrite);
//移除全部写完的BUffer，如果只写了部分数据则更新buffer的readerIndex，下一个writeLoop写入
in.removeBytes(localWrittenBytes);
--writeSpinCount;
break;
}
}
} while (writeSpinCount > 0);
............处理本轮write loop未写完的情况.......
}
case 1 和 default 这两个分支在处理发送数据时的逻辑是一样的，唯一的区别就是 case 1 是处理单个 NioByteBuffer 的发送，而 default 分支是批量处理多个 NioByteBuffers 的发送。
下面笔者就以经常被触发到的 default 分支为例来为大家讲述 Netty 在处理数据发送时的逻辑细节：
-
首先从当前 NioSocketChannel 中的 ChannelOutboundBuffer 中获取本次 write loop 需要发送的字节总量 attemptedBytes 。这个 nioBufferSize 是在前边介绍
ChannelOutboundBuffer#nioBuffers方法转换 JDK NIO ByteBuffer 类型时被计算出来的。 -
调用 JDK NIO 原生 SocketChannel 批量发送 nioBuffers 中的数据。并获取到本次 write loop 一共批量发送了多少字节 localWrittenBytes 。
/**
* @throws NotYetConnectedException
* If this channel is not yet connected
*/
public abstract long write(ByteBuffer[] srcs, int offset, int length)
throws IOException;
-
localWrittenBytes <= 0表示当前 Socket 的写缓存区 SEND_BUF 已满，写不进数据了。那么就需要向当前 NioSocketChannel 对应的 Reactor 注册 OP_WRITE 事件，并停止当前 flush 流程。当 Socket 的写缓冲区有容量可写时，epoll 会通知 reactor 线程继续写入。
protected final void incompleteWrite(boolean setOpWrite) {
// Did not write completely.
if (setOpWrite) {
//这里处理还没写满16次 但是socket缓冲区已满写不进去的情况 注册write事件
//什么时候socket可写了， epoll会通知reactor线程继续写
setOpWrite();
} else {
...........目前还不需要关注这里.......
}
}
向 Reactor 注册 OP_WRITE 事件：
protected final void setOpWrite() {
final SelectionKey key = selectionKey();
if (!key.isValid()) {
return;
}
final int interestOps = key.interestOps();
if ((interestOps & SelectionKey.OP_WRITE) == 0) {
key.interestOps(interestOps | SelectionKey.OP_WRITE);
}
}
关于通过位运算来向 IO 事件集合 interestOps 添加监听 IO 事件的用法，在前边的文章中，笔者已经多次介绍过了，这里不再重复。
- 根据本次 write loop 向 Socket 写缓冲区写入数据的情况，来调整下次 write loop 最大写入字节数。maxBytesPerGatheringWrite 决定每次 write loop 可以从 channelOutboundBuffer 中最多获取多少发送数据。初始值为
SO_SNDBUF大小 * 2 = 293976 = 146988 << 1，最小值为 2048。
public static final int MAX_BYTES_PER_GATHERING_WRITE_ATTEMPTED_LOW_THRESHOLD = 4096;
private void adjustMaxBytesPerGatheringWrite(int attempted, int written, int oldMaxBytesPerGatheringWrite) {
if (attempted == written) {
if (attempted << 1 > oldMaxBytesPerGatheringWrite) {
((NioSocketChannelConfig) config).setMaxBytesPerGatheringWrite(attempted << 1);
}
} else if (attempted > MAX_BYTES_PER_GATHERING_WRITE_ATTEMPTED_LOW_THRESHOLD && written < attempted >>> 1) {
((NioSocketChannelConfig) config).setMaxBytesPerGatheringWrite(attempted >>> 1);
}
}
由于操作系统会动态调整 SO_SNDBUF 的大小，所以这里 netty 也需要根据操作系统的动态调整做出相应的调整，目的是尽量多的去写入数据。
attempted == written 表示本次 write loop 尝试写入的数据能全部写入到 Socket 的写缓冲区中，那么下次 write loop 就应该尝试去写入更多的数据。
那么这里的更多具体是多少呢？
Netty 会将本次写入的数据量 written 扩大两倍，如果扩大两倍后的写入量大于本次 write loop 的最大限制写入量 maxBytesPerGatheringWrite，说明用户的写入需求很猛烈，Netty当然要满足这样的猛烈需求，那么就将当前 NioSocketChannelConfig 中的 maxBytesPerGatheringWrite 更新为本次 write loop 两倍的写入量大小。
在下次 write loop 写入数据的时候，就会尝试从 ChannelOutboundBuffer 中加载最多 written * 2 大小的字节数。
如果扩大两倍后的写入量依然小于等于本次 write loop 的最大限制写入量 maxBytesPerGatheringWrite，说明用户的写入需求还不是很猛烈，Netty 继续维持本次 maxBytesPerGatheringWrite 数值不变。
如果本次写入的数据还不及尝试写入数据的 1 / 2 ：written < attempted >>> 1。说明当前 Socket 写缓冲区的可写容量不是很多了，下一次 write loop 就不要写这么多了尝试减少下次写入的量将下次 write loop 要写入的数据减小为 attempted 的1 / 2。当然也不能无限制的减小，最小值不能低于 2048。
这里可以结合笔者前边的文章《一文聊透ByteBuffer动态自适应扩缩容机制》中介绍到的 read loop 场景中的扩缩容一起对比着看。
read loop 中的扩缩容触发时机是在一个完整的 read loop 结束时候触发。而 write loop 中扩缩容的触发时机是在每次 write loop 发送完数据后，立即触发扩缩容判断。
- 当本次 write loop 批量发送完 ChannelOutboundBuffer 中的数据之后，最后调用
in.removeBytes(localWrittenBytes)从 ChannelOutboundBuffer 中移除全部写完的 Entry ，如果只发送了 Entry 的部分数据则更新 Entry 对象中封装的 DirectByteBuffer 的 readerIndex，等待下一次 write loop 写入。
到这里，write loop 中的发送数据的逻辑就介绍完了，接下来 Netty 会在 write loop 中循环地发送数据直到写满 16 次或者数据发送完毕。
还有一种退出 write loop 的情况就是当 Socket 中的写缓冲区满了，无法在写入时。Netty 会退出 write loop 并向 reactor 注册 OP_WRITE 事件。
但这其中还隐藏着一种情况就是如果 write loop 已经写满 16 次但还没写完数据并且此时 Socket 写缓冲区还没有满，还可以继续在写。那 Netty 会如何处理这种情况呢？
6. 处理Socket可写但已经写满16次还没写完的情况
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
SocketChannel ch = javaChannel();
int writeSpinCount = config().getWriteSpinCount();
do {
.........将待发送数据转换到JDK NIO ByteBuffer中.........
int nioBufferCnt = in.nioBufferCount();
switch (nioBufferCnt) {
case 0:
//这里主要是针对 网络传输文件数据 的处理 FileRegion
writeSpinCount -= doWrite0(in);
break;
case 1: {
.....发送单个nioBuffer....
}
default: {
.....批量发送多个nioBuffers......
}
}
} while (writeSpinCount > 0);
//处理write loop结束 但数据还没写完的情况
incompleteWrite(writeSpinCount < 0);
}
当 write loop 结束后，这时 writeSpinCount 的值会有两种情况：
-
writeSpinCount < 0：这种情况有点不好理解，我们在介绍 Netty 通过零拷贝的方式传输网络文件也就是这里的 case 0 分支逻辑时，详细介绍了 doWrite0 方法的几种返回值，当 Netty 在传输文件的过程中发现 Socket 缓冲区已满无法在继续写入数据时，会返回WRITE_STATUS_SNDBUF_FULL = Integer.MAX_VALUE，这就使得writeSpinCount的值 < 0。随后 break 掉 write loop 来到incompleteWrite(writeSpinCount < 0)方法中，最后会在 incompleteWrite 方法中向 reactor 注册 OP_WRITE 事件。当 Socket 缓冲区变得可写时，epoll 会通知 reactor 线程继续发送文件。
protected final void incompleteWrite(boolean setOpWrite) {
// Did not write completely.
if (setOpWrite) {
//这里处理还没写满16次 但是socket缓冲区已满写不进去的情况 注册write事件
// 什么时候socket可写了， epoll会通知reactor线程继续写
setOpWrite();
} else {
..............
}
}
-
writeSpinCount == 0： 这种情况很好理解，就是已经写满了 16 次，但是还没写完，同时 Socket 的写缓冲区未满，还可以继续写入。这种情况下即使 Socket 还可以继续写入，Netty 也不会再去写了，因为执行 flush 操作的是 reactor 线程，而 reactor 线程负责执行注册在它上边的所有 channel 的 IO 操作，Netty 不会允许 reactor 线程一直在一个 channel 上执行 IO 操作，reactor 线程的执行时间需要均匀的分配到每个 channel 上。所以这里 Netty 会停下，转而去处理其他 channel 上的 IO 事件。
那么还没写完的数据，Netty会如何处理呢？
protected final void incompleteWrite(boolean setOpWrite) {
// Did not write completely.
if (setOpWrite) {
//这里处理还没写满16次 但是socket缓冲区已满写不进去的情况 注册write事件
// 什么时候socket可写了， epoll会通知reactor线程继续写
setOpWrite();
} else {
//这里处理的是socket缓冲区依然可写，但是写了16次还没写完，这时就不能在写了，reactor线程需要处理其他channel上的io事件
//因为此时socket是可写的，必须清除op_write事件，否则会一直不停地被通知
clearOpWrite();
//如果本次writeLoop还没写完，则提交flushTask到reactor
eventLoop().execute(flushTask);
}
这个方法的 if 分支逻辑，我们在介绍do {.....}while()循环体 write loop 中发送逻辑时已经提过，在 write loop 循环发送数据的过程中，如果发现 Socket 缓冲区已满，无法写入数据时（ localWrittenBytes <= 0），则需要向 reactor 注册 OP_WRITE 事件，等到 Socket 缓冲区变为可写状态时，epoll 会通知 reactor 线程继续写入剩下的数据。
do {
.........将待发送数据转换到JDK NIO ByteBuffer中.........
int nioBufferCnt = in.nioBufferCount();
switch (nioBufferCnt) {
case 0:
writeSpinCount -= doWrite0(in);
break;
case 1: {
.....发送单个nioBuffer....
final int localWrittenBytes = ch.write(buffer);
if (localWrittenBytes <= 0) {
incompleteWrite(true);
return;
}
.................省略..............
break;
}
default: {
.....批量发送多个nioBuffers......
final long localWrittenBytes = ch.write(nioBuffers, 0, nioBufferCnt);
if (localWrittenBytes <= 0) {
incompleteWrite(true);
return;
}
.................省略..............
break;
}
}
} while (writeSpinCount > 0);
注意 if 分支处理的情况是还没写满 16 次，但是 Socket 缓冲区已满，无法写入的情况。
而 else 分支正是处理我们这里正在讨论的情况即 Socket 缓冲区是可写的，但是已经写满 16 次，在本轮 write loop 中不能再继续写入的情况。
这时 Netty 会将 channel 中剩下的待写数据的 flush 操作封装程 flushTask，丢进 reactor 的普通任务队列中，等待 reactor 执行完其他 channel 上的 io 操作后在回过头来执行未写完的 flush 任务。
忘记 Reactor 整体运行逻辑的同学，可以在回看下笔者的这篇文章《一文聊透Netty核心引擎Reactor的运转架构》
private final Runnable flushTask = new Runnable() {
@Override
public void run() {
((AbstractNioUnsafe) unsafe()).flush0();
}
};
这里我们看到 flushTask 中的任务是直接再次调用 flush0 继续回到发送数据的逻辑流程中。
细心的同学可能会有疑问，为什么这里不在继续注册 OP_WRITE 事件而是通过向 reactor 提交一个 flushTask 来完成 channel 中剩下数据的写入呢？
原因是这里我们讲的 else 分支是用来处理 Socket 缓冲区未满还是可写的，但是由于用户本次要发送的数据太多，导致写了 16 次还没写完的情形。
既然当前 Socket 缓冲区是可写的，我们就不能注册 OP_WRITE 事件，否则这里一直会不停地收到 epoll 的通知。因为 JDK NIO Selector 默认的是 epoll 的水平触发。
忘记水平触发和边缘触发这两种 epoll 工作模式的同学，可以在回看下笔者的这篇文章《聊聊Netty那些事儿之从内核角度看IO模型》
所以这里只能向 reactor 提交 flushTask 来继续完成剩下数据的写入，而不能注册 OP_WRITE 事件。
注意：只有当 Socket 缓冲区已满导致无法写入时，Netty 才会去注册 OP_WRITE 事件。这和我们之前介绍的 OP_ACCEPT 事件和 OP_READ 事件的注册时机是不同的。
这里大家可能还会有另一个疑问，就是为什么在向 reactor 提交 flushTask 之前需要清理 OP_WRITE 事件呢？ 我们并没有注册 OP_WRITE 事件呀~~
protected final void incompleteWrite(boolean setOpWrite) {
if (setOpWrite) {
......省略......
} else {
clearOpWrite();
eventLoop().execute(flushTask);
}
在为大家解答这个疑问之前，笔者先为大家介绍下 Netty 是如何处理 OP_WRITE 事件的，当大家明白了 OP_WRITE 事件的处理逻辑后，这个疑问就自然解开了。
7. OP_WRITE事件的处理
在《一文聊透Netty核心引擎Reactor的运转架构》一文中，我们介绍过，当 Reactor 监听到 channel 上有 IO 事件发生后，最终会在 processSelectedKey 方法中处理 channel 上的 IO 事件，其中 OP_ACCEPT 事件和 OP_READ 事件的处理过程，笔者已经在之前的系列文章中介绍过了，这里我们聚焦于 OP_WRITE 事件的处理。
public final class NioEventLoop extends SingleThreadEventLoop {
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
.............省略.......
try {
int readyOps = k.readyOps();
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
......处理connect事件......
}
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
ch.unsafe().forceFlush();
}
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
........处理accept和read事件.........
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
}
这里我们看到当 OP_WRITE 事件发生后，Netty 直接调用 channel 的 forceFlush 方法。
@Override
public final void forceFlush() {
// directly call super.flush0() to force a flush now
super.flush0();
}
其实 forceFlush 方法中并没有什么特殊的逻辑，直接调用 flush0 方法再次发起 flush 操作继续 channel 中剩下数据的写入。
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
SocketChannel ch = javaChannel();
int writeSpinCount = config().getWriteSpinCount();
do {
if (in.isEmpty()) {
clearOpWrite();
return;
}
.........将待发送数据转换到JDK NIO ByteBuffer中.........
int nioBufferCnt = in.nioBufferCount();
switch (nioBufferCnt) {
case 0:
......传输网络文件........
case 1: {
.....发送单个nioBuffer....
}
default: {
.....批量发送多个nioBuffers......
}
}
} while (writeSpinCount > 0);
//处理write loop结束 但数据还没写完的情况
incompleteWrite(writeSpinCount < 0);
}
注意这里的 clearOpWrite() 方法，由于 channel 上的 OP_WRITE 事件就绪，表明此时 Socket 缓冲区变为可写状态，从而 Reactor 线程再次来到了 flush 流程中。
当 ChannelOutboundBuffer 中的数据全部写完后 in.isEmpty() ，就需要清理 OP_WRITE 事件，因为此时 Socket 缓冲区是可写的，这种情况下当数据全部写完后，就需要取消对 OP_WRITE 事件的监听，否则 epoll 会不断的通知 Reactor。
同理在 incompleteWrite 方法的 else 分支也需要执行 clearOpWrite() 方法取消对 OP_WRITE 事件的监听。
protected final void incompleteWrite(boolean setOpWrite) {
if (setOpWrite) {
// 这里处理还没写满16次 但是socket缓冲区已满写不进去的情况 注册write事件
// 什么时候socket可写了， epoll会通知reactor线程继续写
setOpWrite();
} else {
// 必须清除OP_WRITE事件，此时Socket对应的缓冲区依然是可写的，只不过当前channel写够了16次，被SubReactor限制了。
// 这样SubReactor可以腾出手来处理其他channel上的IO事件。这里如果不清除OP_WRITE事件，则会一直被通知。
clearOpWrite();
//如果本次writeLoop还没写完，则提交flushTask到SubReactor
//释放SubReactor让其可以继续处理其他Channel上的IO事件
eventLoop().execute(flushTask);
}
}
8. writeAndFlush
在我们讲完了 write 事件和 flush 事件的处理过程之后，writeAndFlush 就变得很简单了，它就是把 write 和 flush 流程结合起来，先触发 write 事件然后在触发 flush 事件。
下面我们来看下 writeAndFlush 的具体逻辑处理：
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
//此处的msg就是Netty在read loop中从NioSocketChannel中读取到ByteBuffer
ctx.writeAndFlush(msg);
}
}
abstract class AbstractChannelHandlerContext implements ChannelHandlerContext, ResourceLeakHint {
@Override
public ChannelFuture writeAndFlush(Object msg) {
return writeAndFlush(msg, newPromise());
}
@Override
public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) {
write(msg, true, promise);
return promise;
}
}
这里可以看到 writeAndFlush 方法的处理入口和 write 事件的处理入口是一样的。唯一不同的是入口处理函数 write 方法的 boolean flush 入参不同，在 writeAndFlush 的处理中 flush = true。
private void write(Object msg, boolean flush, ChannelPromise promise) {
ObjectUtil.checkNotNull(msg, "msg");
................省略检查promise的有效性...............
//flush = true 表示channelHandler中调用的是writeAndFlush方法，这里需要找到pipeline中覆盖write或者flush方法的channelHandler
//flush = false 表示调用的是write方法，只需要找到pipeline中覆盖write方法的channelHandler
final AbstractChannelHandlerContext next = findContextOutbound(flush ?
(MASK_WRITE | MASK_FLUSH) : MASK_WRITE);
//用于检查内存泄露
final Object m = pipeline.touch(msg, next);
//获取下一个要被执行的channelHandler的executor
EventExecutor executor = next.executor();
//确保OutBound事件由ChannelHandler指定的executor执行
if (executor.inEventLoop()) {
//如果当前线程正是channelHandler指定的executor则直接执行
if (flush) {
next.invokeWriteAndFlush(m, promise);
} else {
next.invokeWrite(m, promise);
}
} else {
//如果当前线程不是ChannelHandler指定的executor,则封装成异步任务提交给指定executor执行，注意这里的executor不一定是reactor线程。
final WriteTask task = WriteTask.newInstance(next, m, promise, flush);
if (!safeExecute(executor, task, promise, m, !flush)) {
task.cancel();
}
}
}
由于在 writeAndFlush 流程的处理中，flush 标志被设置为 true，所以这里有两个地方会和 write 事件的处理有所不同。
-
findContextOutbound( MASK_WRITE | MASK_FLUSH )：这里在 pipeline 中向前查找的 ChanneOutboundHandler 需要实现 write 方法或者 flush 方法。这里需要注意的是 write 方法和 flush 方法只需要实现其中一个即可满足查找条件。因为一般我们自定义 ChannelOutboundHandler 时，都会继承 ChannelOutboundHandlerAdapter 类，而在 ChannelInboundHandlerAdapter 类中对于这些 outbound 事件都会有默认的实现。
public class ChannelOutboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelOutboundHandler {
@Skip
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
ctx.write(msg, promise);
}
@Skip
@Override
public void flush(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
}
这样在后面传播 write 事件或者 flush 事件的时候，我们通过上面逻辑找出的 ChannelOutboundHandler 中可能只实现了一个 flush 方法或者 write 方法。不过这样没关系，如果这里在传播 outbound 事件的过程中，发现找出的 ChannelOutboundHandler 中并没有实现对应的 outbound 事件回调函数，那么就直接调用在 ChannelOutboundHandlerAdapter 中的默认实现。
- 在向前传播 writeAndFlush 事件的时候会通过调用 ChannelHandlerContext 的 invokeWriteAndFlush 方法，先传播 write 事件 然后在传播 flush 事件。
void invokeWriteAndFlush(Object msg, ChannelPromise promise) {
if (invokeHandler()) {
//向前传递write事件
invokeWrite0(msg, promise);
//向前传递flush事件
invokeFlush0();
} else {
writeAndFlush(msg, promise);
}
}
private void invokeWrite0(Object msg, ChannelPromise promise) {
try {
//调用当前ChannelHandler中的write方法
((ChannelOutboundHandler) handler()).write(this, msg, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
}
private void invokeFlush0() {
try {
((ChannelOutboundHandler) handler()).flush(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
}
这里我们看到了 writeAndFlush 的核心处理逻辑，首先向前传播 write 事件，经过 write 事件的流程处理后，最后向前传播 flush 事件。
根据前边的介绍，这里在向前传播 write 事件的时候，可能查找出的 ChannelOutboundHandler 只是实现了 flush 方法，不过没关系，这里会直接调用 write 方法在 ChannelOutboundHandlerAdapter 父类中的默认实现。同理 flush 也是一样。
总结
到这里，Netty 处理数据发送的整个完整流程，笔者就为大家详细地介绍完了，可以看到 Netty 在处理读取数据和处理发送数据的过程中，虽然核心逻辑都差不多，但是发送数据的过程明显细节比较多，而且更加复杂一些。
这里笔者将读取数据和发送数据的不同之处总结如下几点供大家回忆对比：
-
在每次 read loop 之前，会分配一个大小固定的 diretByteBuffer 用来装载读取数据。每轮 read loop 完全结束之后，才会决定是否对下一轮的读取过程分配的 directByteBuffer 进行扩容或者缩容。
-
在每次 write loop 之前，都会获取本次 write loop 最大能够写入的字节数，根据这个最大写入字节数从 ChannelOutboundBuffer 中转换 JDK NIO ByteBuffer 。每次写入 Socket 之后都需要重新评估是否对这个最大写入字节数进行扩容或者缩容。
-
read loop 和 write loop 都被默认限定最多执行 16 次。
-
在一个完整的 read loop 中，如果还读取不完数据，直接退出。等到 reactor 线程执行完其他 channel 上的 IO 事件再来读取未读完的数据。
-
而在一个完整的 write loop 中，数据发送不完，则分两种情况。
- Socket 缓冲区满无法在继续写入。这时就需要向 reactor 注册 OP_WRITE 事件。等 Socket 缓冲区变的可写时，epoll 通知 reactor 线程继续发送。
- Socket 缓冲区可写，但是由于发送数据太多，导致虽然写满 16 次但依然没有写完。这时就直接向 reactor 丢一个 flushTask 进去，等到 reactor 线程执行完其他 channel 上的 IO 事件，在回过头来执行 flushTask。
-
OP_READ 事件的注册是在 NioSocketChannel 被注册到对应的 Reactor 中时就会注册。而 OP_WRITE 事件只会在 Socket 缓冲区满的时候才会被注册。当 Socket 缓冲区再次变得可写时，要记得取消 OP_WRITE 事件的监听。否则的话就会一直被通知。
好了，本文的全部内容就到这里了，我们下篇文章见~~~~
网友评论