上一篇文章对RecvByteBufAllocator的源码进行了分析，介绍了Netty是如何接收对端发送过来的数据。以及Netty是如何通过AdaptiveRecvByteBufAllocator来自适应调整ByteBuf的动态分配，解决Java ByteBuffer分配过大浪费内存，分配过小又需要频繁扩容的问题。
本篇文章会分析，Netty是如何将数据发送出去的。
前置知识
Netty支持的数据传输类型
首先你需要知道，通过Netty来发送数据，它只支持两种数据类型：ByteBuf和FileRegion。前者可以看作是ByteBuffer，普通的字节数据传输。而后者是文件传输，Netty通过FileRegion来实现文件传输的零拷贝。
write和flush
write()并不会发送数据，只是简单的将数据暂存到ChannelOutboundBuffer。
flush()才是真正的将数据通过Socket传输给对端。
writeAndFlush()只是简单的执行以上两个方法而已。
Channel高低水位线
当程序write了大量数据，或者虽然调用了flush()，但是由于对端来不及接收数据，再或者由于网络原因等等情况，导致TCP缓冲区被写满，大量的消息积压在ChannelOutboundBuffer，导致内存溢出。为了保护你的程序，Netty给Channel设置了「高低水位线」，当积压的消息超过了高水位，Netty会将Channel设为「不可写」状态并触发channelWritabilityChanged回调，你可以通过Channel.isWritable()判断是否要继续写数据。
通过ChannelConfig.setWriteBufferHighWaterMark()和ChannelConfig.setWriteBufferLowWaterMark()设置Channel的高低水位线。
订阅OP_WRITE事件
既然write()操作是用户自己发起的，为啥还要订阅Channel的OP_WRITE事件呢？
因为TCP缓冲区可能被写满，此时你就应该订阅OP_WRITE事件，暂时放弃写操作，等待Selector通知你Channel可写时，你再继续写。
ByteBuf可以转换成ByteBuffer
Java原生的SocketChannel只支持写入ByteBuffer，当你通过Netty写入ByteBuf时，它会将ByteBuf转换成ByteBuffer再写入，方法是ByteBuf.internalNioBuffer()。
清楚Java对象在JVM中的内存布局
write(msg)时，会将msg包装成Entry节点加入到链尾，其中一个属性pendingSize记录着消息占用的内存空间，这个空间大小除了msg数据本身占用的空间外，还包含Entry对象占用的空间，因此默认会额外再加上96。为啥是96后面会说明，首先你应该知道对象的对象头最大占用16字节，对象引用最少占用4字节，最多占用8字节，一个long类型占用8字节，int类型占用4字节，boolean类型占用1字节。另外JVM要求Java对象占用的空间必须是8字节的整数倍，因此还会有padding填充字节。
ChannelHandlerContext.writeAndFlush()分析
如下，分别是发送一个ByteBuf和FileRegion的简单示例：
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 发送一个 hello
ctx.writeAndFlush(Unpooled.wrappedBuffer("hello".getBytes()));
}
复制代码@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 传输一个 a.txt 文件
RandomAccessFile accessFile = new RandomAccessFile("/disk/a.txt", "r");
DefaultFileRegion region = new DefaultFileRegion(accessFile.getChannel(), 0, accessFile.length());
ctx.writeAndFlush(region);
}
复制代码我先说一下writeAndFlush的整体流程，实际的发送细节下一节会解释。
调用ctx.writeAndFlush()，会从当前Handler往前找能处理write事件的Handler，如果调用的是ctx.channel().writeAndFlush()，则会从Pipeline的TailContext开始向前找能处理write事件的Handler，事件传播的路径稍微有点区别。默认情况下，会找到HeadContext来处理，源码如下：
private void write(Object msg, boolean flush, ChannelPromise promise) {
// 确保发送的消息不为空
ObjectUtil.checkNotNull(msg, "msg");
try {
if (isNotValidPromise(promise, true)) {
ReferenceCountUtil.release(msg);
// cancelled
return;
}
} catch (RuntimeException e) {
ReferenceCountUtil.release(msg);
throw e;
}
// 往后找能处理 write事件的Channel，默认会找到HeadContext。
final AbstractChannelHandlerContext next = findContextOutbound(flush ?
(MASK_WRITE | MASK_FLUSH) : MASK_WRITE);
final Object m = pipeline.touch(msg, next);
EventExecutor executor = next.executor();
// 如果是EventLoop线程，则直接执行，否则提交一个任务串行化执行。
if (executor.inEventLoop()) {
if (flush) {
// 调用的是writeAndFlush()，所有flush为true，这里会调用HeadContext.invokeWriteAndFlush()
next.invokeWriteAndFlush(m, promise);
} else {
next.invokeWrite(m, promise);
}
} else {
final WriteTask task = WriteTask.newInstance(next, m, promise, flush);
if (!safeExecute(executor, task, promise, m, !flush)) {
task.cancel();
}
}
}
复制代码找到HeadContext后，调用其invokeWriteAndFlush()方法，其实就是将write和flush放在一个方法里调用了：
void invokeWriteAndFlush(Object msg, ChannelPromise promise) {
if (invokeHandler()) {
// 先通过handler调用write()
invokeWrite0(msg, promise);
// 再通过handler调用flush()
invokeFlush0();
} else {
writeAndFlush(msg, promise);
}
}
复制代码先看invokeWrite0()，它会调用HeadContext.write()，由于write操作需要和JDK的底层API交互，于是操作又会被转交给Channel.Unsafe执行：
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
// 需要和JDK底层API交互，转交给Unsafe执行。
unsafe.write(msg, promise);
}
复制代码接下来会调用AbstractChannel.AbstractUnsafe.write()方法，它首先会对发送的数据做过滤，只支持ByteBuf和FileRegion两种类型。然后会计算发送的数据占用的内存大小，因为前面说过积压的消息一旦超过Channel的高水位线会将Channel设为「不可写」状态，防止内存溢出。这两步做完以后，会把消息添加到输出缓冲区ChannelOutboundBuffer中。
@Override
public final void write(Object msg, ChannelPromise promise) {
assertEventLoop();
ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
if (outboundBuffer == null) {// outboundBuffer会随着Channel一同被创建，一般不会为null，这里做了校验。
try {
ReferenceCountUtil.release(msg);
} finally {
safeSetFailure(promise,
newClosedChannelException(initialCloseCause, "write(Object, ChannelPromise)"));
}
return;
}
int size;
try {
// 过滤写出消息，确保是ByteBuf或FileRegion，其他对象不支持写出。
msg = filterOutboundMessage(msg);
/*
估算消息占用的内存，作用:
因为write()不会把消息写出到Socket，会暂存在内存里，直到flush()。
Netty为了防止消息堆积，会设置高低水位，消息暂存的总量达到最高水位会将Channel设置不可写状态，
以保护你的程序，避免内存溢出。
详见:io.netty.channel.DefaultMessageSizeEstimator.HandleImpl.size()
对于FileRegion，会直接返回0，因为使用了零拷贝技术，不需要把文件读取到JVM进程。
*/
size = pipeline.estimatorHandle().size(msg);
if (size < 0) {
size = 0;
}
} catch (Throwable t) {
try {
ReferenceCountUtil.release(msg);
} finally {
safeSetFailure(promise, t);
}
return;
}
// write()只会把消息暂存在outboundBuffer，不会真正发送。
outboundBuffer.addMessage(msg, size, promise);
}
复制代码关注一下filterOutboundMessage()，它除了过滤消息，还会试图将HeapByteBuf转换成DirectByteBuf。Netty为了提升数据发送的效率，和Socket直接读写的数据会使用直接内存，避免IO操作再发生内存拷贝。
// 过滤出站消息，只支持写出ByteBuf和FileRegion。
@Override
protected final Object filterOutboundMessage(Object msg) {
if (msg instanceof ByteBuf) {
ByteBuf buf = (ByteBuf) msg;
if (buf.isDirect()) {
return msg;
}
// 为了避免内存复制，Socket直接读写的数据都使用堆外内存
return newDirectBuffer(buf);
}
// 文件传输
if (msg instanceof FileRegion) {
return msg;
}
// 不支持的数据类型，抛异常
throw new UnsupportedOperationException(
"unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES);
}
复制代码newDirectBuffer()并不保证一定转换成功，如果使用的ByteBufAllocator是未池化的，且没有开启io.netty.threadLocalDirectBufferSize，那么就意味着Netty需要申请一个没有被池化的DirectByteBuf，这个操作是非常昂贵的，Netty会放弃转换：
// 试图将HeapByteBuf转换成DirectByteBuf，如果转换的开销很大会放弃。
protected final ByteBuf newDirectBuffer(ByteBuf buf) {
final int readableBytes = buf.readableBytes();
if (readableBytes == 0) {
// 可读字节数为0，直接释放并返回共享空对象。
ReferenceCountUtil.safeRelease(buf);
return Unpooled.EMPTY_BUFFER;
}
// 获取Channel绑定的ByteBufAllocator
final ByteBufAllocator alloc = alloc();
if (alloc.isDirectBufferPooled()) {// 分配器是否是池化的，且能分配直接内存？
// 创建一个指定大小的直接内存ByteBuf，将数据写入，原buf释放
ByteBuf directBuf = alloc.directBuffer(readableBytes);
directBuf.writeBytes(buf, buf.readerIndex(), readableBytes);
ReferenceCountUtil.safeRelease(buf);
return directBuf;
}
/*
如果设置了io.netty.threadLocalDirectBufferSize，Netty会在线程的FastThreadLocal中通过Stack实现一个轻量级的
ByteBuf对象池，ByteBuf写出到Socket后，会自动释放，这里会将它再push到线程绑定的Stack中进行重用。
*/
final ByteBuf directBuf = ByteBufUtil.threadLocalDirectBuffer();
if (directBuf != null) {
directBuf.writeBytes(buf, buf.readerIndex(), readableBytes);
ReferenceCountUtil.safeRelease(buf);
return directBuf;
}
// 申请一个未池化的直接内存ByteBuf开销是很大的，测试过，比堆内存的申请慢10倍都不止，这里会直接放弃。
return buf;
}
复制代码如果设置了io.netty.threadLocalDirectBufferSize，Netty会为每个线程创建指定数量的ByteBuf对象缓存，这些ByteBuf是可以被重用的。实现逻辑是Netty会在FastThreadLocal存放一个Stack，需要时pop()一个出来，用完时push()归还。
再来关注一下MessageSizeEstimator，它负责计算待发送数据占用的内存，逻辑很简单，对于FileRegion会返回0，因为FileRegion传输文件时使用了零拷贝技术，直接使用mmap内存映射，而不需要将文件加载到JVM进程，实现直接看io.netty.channel.DefaultMessageSizeEstimator.HandleImpl.size()：
// 估算消息的内存占用，逻辑还是很简单的。
@Override
public int size(Object msg) {
if (msg instanceof ByteBuf) {
return ((ByteBuf) msg).readableBytes();
}
if (msg instanceof ByteBufHolder) {
return ((ByteBufHolder) msg).content().readableBytes();
}
// FileRegion实现了零拷贝，并不需要将文件加载到JVM，因此占用的内存为0，不影响Channel水位线。
if (msg instanceof FileRegion) {
return 0;
}
return unknownSize;
}
复制代码关于ChannelOutboundBuffer代码下节会详细分析，这里只需要知道write()只会将数据暂存到ChannelOutboundBuffer，而不会真正发送就行了。
消息存入ChannelOutboundBuffer，write操作就算完成了。紧接着会调用invokeFlush0()，它依然会转交给Unsafe执行，调用AbstractChannel.AbstractUnsafe.flush()。
它会做两件事：先把ChannelOutboundBuffer中待发送的Entry标记为flushed，然后将要发送的Entry数据转换成Java的ByteBuffer，使用SocketChannel进行真正的数据发送。
@Override
public final void flush() {
assertEventLoop();
// 得到SocketChannel绑定的ChannelOutboundBuffer
ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
if (outboundBuffer == null) {
return;
}
// 先将unflushed节点标记为flushed
outboundBuffer.addFlush();
// 开始发送数据
flush0();
}
复制代码flush0()会开始发送数据，它首先会检测Channel是否活跃，如果是非活跃状态，此次flush()操作将失败，Entry会被移除。如果Channel正常，会调用doWrite()进行数据发送。
protected void flush0() {
if (inFlush0) {// 避免上一次flush0()还没执行完时，又触发了
return;
}
final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
if (outboundBuffer == null || outboundBuffer.isEmpty()) {// 非空校验
return;
}
inFlush0 = true;
// 如果连接已经失活了。
if (!isActive()) {
try {
if (!outboundBuffer.isEmpty()) {
if (isOpen()) {
/*
通道是打开的，稍后可能会被激活。
1.释放msg
2.触发失败通知
3.回收Entry
4.递减消息挂起的字节数
*/
outboundBuffer.failFlushed(new NotYetConnectedException(), true);
} else {
/*
道都被关闭了，和上面的处理流程类似，只是不用通过触发channelWritabilityChanged()回调了。
*/
outboundBuffer.failFlushed(newClosedChannelException(initialCloseCause, "flush0()"), false);
}
}
} finally {
inFlush0 = false;
}
return;
}
try {
// 连接正常，执行真正的write()操作
doWrite(outboundBuffer);
} catch (Throwable t) {
handleWriteError(t);
} finally {
inFlush0 = false;
}
}
复制代码doWrite()是数据发送的核心，由子类实现，这里直接看NioSocketChannel.doWrite()。它会获取Java原生的SocketChannel，将队列中待发送的ByteBuf转换成ByteBuffer，然后循环发送数据。
单次循环发送的数据量受以下两个条件限制：
- ByteBuffer的数量限制。
- TCP参数设置的缓冲区的大小限制(ChannelOption.SO_SNDBUF)。
如果ChannelOutboundBuffer积压了大量的数据，单次可能无法发送完，因此会默认循环发16次。循环次数过多可能会阻塞IO线程，导致其他Channel的事件得不到处理。
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
SocketChannel ch = javaChannel();
// 循环写的次数，默认16次。可能有大量消息积压在输出缓冲区，同时为了避免阻塞IO线程，做了次数限制。
int writeSpinCount = config().getWriteSpinCount();
do {
if (in.isEmpty()) {// 没有数据要写出了
// 取消监听 OP_WRITE事件
clearOpWrite();
return;
}
// 发送缓冲区的最大值，由 TCP参数: ChannelOption.SO_SNDBUF 设置。
int maxBytesPerGatheringWrite = ((NioSocketChannelConfig) config).getMaxBytesPerGatheringWrite();
/*
将要flush的Entry转换成Java原生的ByteBuffer数组，因为做了总数和总字节数的限制，所以一次可能无法send所有数据。
注意，这里只会处理ByteBuf，不会处理FileRegion。
*/
ByteBuffer[] nioBuffers = in.nioBuffers(1024, maxBytesPerGatheringWrite);
// 其实就是 nioBuffers.length，上一步方法中会进行设置
int nioBufferCnt = in.nioBufferCount();
switch (nioBufferCnt) {
case 0:
// ByteBuf处理完了，但是可能还有FileRegion需要处理。
writeSpinCount -= doWrite0(in);
break;
case 1: {
// 只有单个ByteBuf需要发送的情况
ByteBuffer buffer = nioBuffers[0];
// 尝试发送的字节数
int attemptedBytes = buffer.remaining();
// Java原生的SocketChannel.wrote(ByteBuffer)发送数据
final int localWrittenBytes = ch.write(buffer);
if (localWrittenBytes <= 0) {// TCP缓冲区满，订阅OP_WRITE事件，等待可写时再继续处理
incompleteWrite(true);
return;
}
// 动态调整 发送缓冲区大小
adjustMaxBytesPerGatheringWrite(attemptedBytes, localWrittenBytes, maxBytesPerGatheringWrite);
// 删除已经发送的Entry节点
in.removeBytes(localWrittenBytes);
--writeSpinCount;
break;
}
default: {
// 发送缓冲区有多个ByteBuf待发送
// 尝试发送的字节总数
long attemptedBytes = in.nioBufferSize();
// 调用Java原生的SocketChannel.write()进行数据的发送，返回值是实际发送的字节数
final long localWrittenBytes = ch.write(nioBuffers, 0, nioBufferCnt);
if (localWrittenBytes <= 0) {
// 写入字节为0，可能是TCP缓冲区满了，订阅OP_WRITE事件，等待TCP可写时再执行。
incompleteWrite(true);
return;
}
// 根据本次实际写入的字节数，动态调整发送缓冲区:ChannelOption.SO_SNDBUF
adjustMaxBytesPerGatheringWrite((int) attemptedBytes, (int) localWrittenBytes,
maxBytesPerGatheringWrite);
/*
删除已经发送的数据，是根据实际写入的字节数去删除的，而不是根据ByteBuf的数量。
从flushedEntry开始，计算每个ByteBuf的大小，按个删除。
可能存在某个ByteBuf发送部分数据的情况，会调整它的readerIndex。
*/
in.removeBytes(localWrittenBytes);
--writeSpinCount;
break;
}
}
} while (writeSpinCount > 0);
/*
只有在 nioBufferCnt处理完了，调用doWrite0(in)处理FileRegion，且没处理完毕时，才可能走到这里。
如果FileRegion没有处理完，writeSpinCount会小于0，这里会继续订阅OP_WRITE事件，等待Channel可写时继续处理。
*/
incompleteWrite(writeSpinCount < 0);
}
复制代码此外，NioSocketChannel.doWrite()只会发送ByteBuf，FileRegion的发送需要调用父类的AbstractNioByteChannel.doWrite0()处理。
/*
NioSocketChannel只负责发送ByteBuf，
FileRegion的发送这边会处理。
*/
protected final int doWrite0(ChannelOutboundBuffer in) throws Exception {
Object msg = in.current();
if (msg == null) {
// Directly return here so incompleteWrite(...) is not called.
return 0;
}
// 数据发送
return doWriteInternal(in, in.current());
}
private int doWriteInternal(ChannelOutboundBuffer in, Object msg) throws Exception {
if (msg instanceof ByteBuf) {
ByteBuf buf = (ByteBuf) msg;
if (!buf.isReadable()) {
// 没有数据可读，直接删除节点
in.remove();
return 0;
}
// Java底层write()
final int localFlushedAmount = doWriteBytes(buf);
if (localFlushedAmount > 0) {
in.progress(localFlushedAmount);
if (!buf.isReadable()) {
in.remove();
}
return 1;
}
} else if (msg instanceof FileRegion) {
FileRegion region = (FileRegion) msg;
// 已经传输的字节数 >= 字节总数，代表文件已经传输完毕，删除节点。
if (region.transferred() >= region.count()) {
in.remove();
return 0;
}
// 调用region.transferTo(javaChannel(), position)文件传输
long localFlushedAmount = doWriteFileRegion(region);
if (localFlushedAmount > 0) {// 实际发送的字节数
in.progress(localFlushedAmount);
if (region.transferred() >= region.count()) {//FileRegion发送完毕，移除节点
in.remove();
}
return 1;
}
} else {
// Should not reach here.
throw new Error();
}
/*
一般不会走到这里，可能是数据没有处理完毕，返回一个Integer.MAX_VALUE，
让writeSpinCount小于0，这样它就会订阅OP_WRITE事件，等待Channel可写时继续处理。
*/
return WRITE_STATUS_SNDBUF_FULL;
}
复制代码需要注意的是，flush()操作可能存在两种情况：
- 数据正常发送完毕。
- 数据没有发完，就已经超过了最大循环次数，为了不阻塞IO线程，下次再处理。
- TCP缓冲区满，数据无法发送。
对于后面两种情况，都属于「不完整写入」，因此会调用incompleteWrite(setOpWrite)稍后继续处理。
针对第三种情况，Netty需要订阅OP_WRITE事件，等待Selector通知Channel可写时继续发送数据。setOpWrite参数代表是否要监听OP_WRITE事件：
/**
* 不完整写入
* @param setOpWrite 是否要订阅OP_WRITE事件
*/
protected final void incompleteWrite(boolean setOpWrite) {
// setOpWrite为true，一般都是TCP缓冲区满了，此时需要订阅OP_WRITE事件，等待Channel可写时再继续处理。
if (setOpWrite) {
// 订阅OP_WRITE事件
setOpWrite();
} else {
// 取消订阅OP_WRITE事件
clearOpWrite();
// 提交一个flush任务，稍后执行，避免阻塞IO线程。
eventLoop().execute(flushTask);
}
}
复制代码至此，writeAndFlush()的整个流程就处理完了，对于ChannelOutboundBuffer本节没有进行分析，看下节。
ChannelOutboundBuffer源码分析
ChannelOutboundBuffer是Netty的数据发送缓冲区，它跟随SocketChannel一同被创建。
先看属性：
/*
将ByteBuf包装成Entry时，额外占用的字节大小，因为除了ByteBuf本身的数据外，Entry对象也占用空间。
为啥是96？为啥还支持修改？？
1.96是Netty根据64位的JVM计算的最大值。
2.如果你的程序运行在32位的JVM中，或者对象引用开启了压缩，你可以根据实际情况修改这个值。
分析为啥最多会占用96字节：
在64位的JVM中，一个Entry对象占用以下空间:
- 16字节的对象头空间
- 6个对象引用属性，最小4*6=24字节，最大8*6=48字节
- 2个long属性，2*8=16字节
- 2个int属性，2*4=8字节
- 1个boolean属性，1字节
- padding对齐填充，JVM要求对象占用的内存为8字节的整数倍，这里是7字节
合计最多为96字节.
*/
static final int CHANNEL_OUTBOUND_BUFFER_ENTRY_OVERHEAD =
SystemPropertyUtil.getInt("io.netty.transport.outboundBufferEntrySizeOverhead", 96);
/*
发送数据时，需要将待发送的ByteBuf转换成ByteBuffer，考虑到write是个很频繁的操作，
为了避免频繁创建数组，这里进行了复用，每个线程会复用自己的ByteBuffer[]。
*/
private static final FastThreadLocal<ByteBuffer[]> NIO_BUFFERS = new FastThreadLocal<ByteBuffer[]>() {
@Override
protected ByteBuffer[] initialValue() throws Exception {
// 默认大小为1024，后面有必要还会扩容
return new ByteBuffer[1024];
}
};
// 绑定的SocketChannel
private final Channel channel;
// 已经flush，等待发送的头节点。
private Entry flushedEntry;
// 已经write但是没flush的头节点，flush()时会通过它一直往后找
private Entry unflushedEntry;
// 链尾节点
private Entry tailEntry;
// flush的节点数量，发送数据时会从flushedEntry开始往后找flushed个节点。
private int flushed;
// 单次循环写的Nio Buffer数量
private int nioBufferCount;
// 单次循环写的Nio Buffer总大小
private long nioBufferSize;
// flush是否失败
private boolean inFail;
// 计算totalPendingSize属性的偏移量，通过CAS的方式来做修改。
private static final AtomicLongFieldUpdater<ChannelOutboundBuffer> TOTAL_PENDING_SIZE_UPDATER =
AtomicLongFieldUpdater.newUpdater(ChannelOutboundBuffer.class, "totalPendingSize");
// 输出缓冲区暂存的消息占用的总内存，通过该值判断是否达到高低水位，以修改Channel的可写状态。
@SuppressWarnings("UnusedDeclaration")
private volatile long totalPendingSize;
// unwritable属性的偏移量，通过CAS的方式来修改。
private static final AtomicIntegerFieldUpdater<ChannelOutboundBuffer> UNWRITABLE_UPDATER =
AtomicIntegerFieldUpdater.newUpdater(ChannelOutboundBuffer.class, "unwritable");
// Channel是否可写的状态，0可写，1不可写。输出缓冲区内存达到高低水位线时修改。
@SuppressWarnings("UnusedDeclaration")
private volatile int unwritable;
// Channel可写状态变更时触发的任务，消息的积压达到高低水位线时触发
private volatile Runnable fireChannelWritabilityChangedTask;
复制代码它本身是一个单向链表，由一系列Entry节点组成。它有三个节点指针：
- flushedEntry：已经flush，等待被发送的起始节点指针。
- unflushedEntry：已经write，等待flush的起始节点指针。
- tailEntry：链尾指针。
笔者花了一个简图，来表示它是如何工作的：
上节说过，执行flush(msg)操作时，只是把数据暂存到ChannelOutboundBuffer，核心方法是addMessage()，它主要做了两件事：
- 将msg封装成Entry节点，加入到链尾。
- 统计输出缓冲区的消息总字节数是否达到高水位线，如果达到则将Channel设为「不可写」状态，且触发
ChannelWritabilityChanged回调。
/**
* 将消息暂存到ChannelOutboundBuffer，暂存成功promise就会收到通知。
* @param msg 待发送的数据:ByteBuf/FileRegion
* @param size 数据占用的内存大小
* @param promise write成功会收到通知
*/
public void addMessage(Object msg, int size, ChannelPromise promise) {
// 将msg包装成一个Entry，并加入到链尾。
Entry entry = Entry.newInstance(msg, size, total(msg), promise);
if (tailEntry == null) {
flushedEntry = null;
} else {
// tail不为空，则添加到它的next
Entry tail = tailEntry;
tail.next = entry;
}
tailEntry = entry;// tailEntry指向新添加的节点
if (unflushedEntry == null) {
unflushedEntry = entry;
}
// 统计消息挂起的字节数，如果超过高水位线，需要修改Channel的可写状态并触发回调
incrementPendingOutboundBytes(entry.pendingSize, false);
}
复制代码先看Entry.newInstance()，它会将msg封装成Entry节点，加入到链尾。
Entry有个属性pendingSize用来记录消息占用的内存空间，需要注意的是，它除了msg本身的数据空间，还会加上Entry对象占用的空间，一个Java对象占用多少空间是在编译期就确定下来的，除了属性占用的空间外，读者还需要了解Java对象的内存布局。
/**
* 创建一个Entry节点，从对象池中取一个
* @param msg 消息本身
* @param size 由MessageSizeEstimator估算出消息占用的内存大小
* @param total 消息本身的大小(区别是FileRegion的处理)
* @param promise write()完成会收到通知
* @return
*/
static Entry newInstance(Object msg, int size, long total, ChannelPromise promise) {
/*
每次write()都需要一个Entry，考虑到write()是一个非常频繁的操作，
为了避免Entry的频繁创建和销毁，这里做了对象池的重用处理。
*/
Entry entry = RECYCLER.get();
entry.msg = msg;
/*
占用的内存为什么还要加上CHANNEL_OUTBOUND_BUFFER_ENTRY_OVERHEAD？？？
除了ByteBuf占用的空间外，Entry本身也占用空间啊。
在64位的JVM中:
- 16字节的对象头空间
- 6个对象引用属性，最小4*6=24字节，最大8*6=48字节
- 2个long属性，2*8=16字节
- 2个int属性，2*4=8字节
- 1个boolean属性，1字节
- padding对齐填充，JVM要求对象占用的内存为8字节的整数倍，这里是7字节
合计最多为96字节，因此CHANNEL_OUTBOUND_BUFFER_ENTRY_OVERHEAD的默认值即使96。
*/
entry.pendingSize = size + CHANNEL_OUTBOUND_BUFFER_ENTRY_OVERHEAD;
entry.total = total;
entry.promise = promise;
return entry;
}
复制代码Entry加入链表后，incrementPendingOutboundBytes()会累加字节总数，判断是否超过高水位线：
/**
* 统计消息挂起的字节数，如果超过高水位线，需要修改Channel的可写状态并触发回调
* @param size 消息占用内存的字节数
* @param invokeLater 是否稍后触发再回调
*/
private void incrementPendingOutboundBytes(long size, boolean invokeLater) {
if (size == 0) {
return;
}
// 统计消息暂存的内存大小，CAS的方式进行累加
long newWriteBufferSize = TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, size);
// 暂存消息达到高水位线，消息积压了
if (newWriteBufferSize > channel.config().getWriteBufferHighWaterMark()) {
// 修改可写状态，触发回调
setUnwritable(invokeLater);
}
}
复制代码setUnwritable()会在数据总字节数超过高水位线时触发，它会通过自旋+CAS的方式将unwritable从0改为1，然后触发回调：
// 将Channel设为不可写，CAS执行
private void setUnwritable(boolean invokeLater) {
for (;;) {
final int oldValue = unwritable;
final int newValue = oldValue | 1;
if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
if (oldValue == 0) {
// CAS操作成功，触发回调
fireChannelWritabilityChanged(invokeLater);
}
break;
}
}
}
复制代码ChannelOutboundBuffer为write操作所做的工作就这么多了，下面看flush。
执行flush操作时，首先会调用outboundBuffer.addFlush()将unflushed节点标记为flushed，其实就是移动flushedEntry和unflushedEntry指针，这个过程会检查Entry节点是否被取消，如果取消了会跳过节点，同时会递减该Entry占用的内存空间。
// 只是将节点标记为flushed，并没有真正发送数据，会跳过已经被取消的节点。
public void addFlush() {
Entry entry = unflushedEntry;
if (entry != null) {
if (flushedEntry == null) {
flushedEntry = entry;
}
do {
flushed ++;
// 将entry的promise设为 不可取消 状态
if (!entry.promise.setUncancellable()) {
// 设置失败，说明promise已经取消，需要释放消息，并递减挂起的字节数
int pending = entry.cancel();
// 递减缓冲区的消息字节总数，如果达到低水位线，则将Channel重新设为「可写」状态，并触发回调
decrementPendingOutboundBytes(pending, false, true);
}
entry = entry.next;
} while (entry != null);// 不断往后找 待flush的节点
// 所有的节点都flush了，置为空
unflushedEntry = null;
}
}
复制代码节点状态标记完成后，会调用doWrite()开始写数据。首先它需要ChannelOutboundBuffer将flushed节点转换成Java原生的ByteBuffer，方法是nioBuffers()。因为OS对SocketChannel.write()单次发送的字节数有限制，一般是Integer.MAX_VALUE，所以单次转换需要提供两个参数：
- maxCount：转换ByteBuffer的最大数量，默认是1024。
- maxBytes：最大字节数，默认是设置的TCP发送缓冲区大小（ChannelOption.SO_SNDBUF）。
/**
* 将要flush的Entry转换成Java原生的ByteBuffer数组，因为做了总数和总字节数的限制，所以一次可能无法send所有数据。
* @param maxCount 单次发送的ByteBuffer最大数量
* @param maxBytes 发送缓冲区的最大值，由 TCP参数: ChannelOption.SO_SNDBUF 设置。
* @return
*/
public ByteBuffer[] nioBuffers(int maxCount, long maxBytes) {
assert maxCount > 0;
assert maxBytes > 0;
long nioBufferSize = 0;
int nioBufferCount = 0;
// 由于write操作很频繁，避免ByteBuffer[]频繁创建和销毁，这里进行了复用，每个线程都有一个ByteBuffer[1024]
final InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.get();
ByteBuffer[] nioBuffers = NIO_BUFFERS.get(threadLocalMap);
Entry entry = flushedEntry;
// 确保Entry节点是flushed，且msg是ByteBuf类型
while (isFlushedEntry(entry) && entry.msg instanceof ByteBuf) {
// 确保节点没有被取消，如果取消了，则跳过它。
if (!entry.cancelled) {
ByteBuf buf = (ByteBuf) entry.msg;
final int readerIndex = buf.readerIndex();
final int readableBytes = buf.writerIndex() - readerIndex;
// 可读字节数就是要写出的字节数，确保大于0
if (readableBytes > 0) {
if (maxBytes - readableBytes < nioBufferSize && nioBufferCount != 0) {
// 发送的数据超过了maxBytes，退出循环
break;
}
nioBufferSize += readableBytes;
int count = entry.count;
if (count == -1) {
// -1代表没有设置ByteBuf的nioBufferCount，ByteBuf中ByteBuffer的数量
entry.count = count = buf.nioBufferCount();
}
// 是否需要更多的空间
int neededSpace = min(maxCount, nioBufferCount + count);
// 如果ByteBuffer的数量超过了默认值1024，就去申请更多的空间
if (neededSpace > nioBuffers.length) {
// 成倍扩容，直到数组长度足够。并塞回到FastThreadLocal。
nioBuffers = expandNioBufferArray(nioBuffers, neededSpace, nioBufferCount);
NIO_BUFFERS.set(threadLocalMap, nioBuffers);
}
if (count == 1) {
ByteBuffer nioBuf = entry.buf;
if (nioBuf == null) {
// 将ByteBuf转ByteBuffer，且缓存到Entry中
entry.buf = nioBuf = buf.internalNioBuffer(readerIndex, readableBytes);
}
// 设置ByteBuffer
nioBuffers[nioBufferCount++] = nioBuf;
} else {
// 一个ByteBuf包含多个ByteBuffer的情况处理，循环遍历设置
nioBufferCount = nioBuffers(entry, buf, nioBuffers, nioBufferCount, maxCount);
}
if (nioBufferCount >= maxCount) {
// ByteBuffer的数量超过了maxCount，退出循环
break;
}
}
}
entry = entry.next;
}
this.nioBufferCount = nioBufferCount;
this.nioBufferSize = nioBufferSize;
return nioBuffers;
}
复制代码这里关注一下NIO_BUFFERS属性，它是一个FastThreadLocal，每个线程都有自己的ByteBuffer[]缓存，默认长度1024，可以被复用。
这里为什么要复用呢？因为作为一个网络IO框架，flush肯定是一个非常频繁的操作，为了避免每次都创建ByteBuffer[]，复用可以提升系统性能，减轻GC的压力。
如果一个ByteBuf由很多个ByteBuffer组成，默认的1024个ByteBuffer可能不够用，此时会调用expandNioBufferArray()进行扩容：
// 对array进行扩容
private static ByteBuffer[] expandNioBufferArray(ByteBuffer[] array, int neededSpace, int size) {
int newCapacity = array.length;
do {
// 成倍扩容
newCapacity <<= 1;
if (newCapacity < 0) {// int溢出
throw new IllegalStateException();
}
} while (neededSpace > newCapacity);
ByteBuffer[] newArray = new ByteBuffer[newCapacity];
// 元素迁移
System.arraycopy(array, 0, newArray, 0, size);
return newArray;
}
复制代码将待发送的ByteBuf转换成ByteBuffer后，NioSocketChannel会调用JDK底层的SocketChannel.write()进行真正的数据发送。
数据发送完毕后，需要移除ChannelOutboundBuffer中的节点。节点的添加是从链尾开始，移除则是从链头开始的。
ChannelOutboundBuffer是根据实际发送的字节数来移除节点的，因此会存在某个ByteBuf只发送了部分数据的情况，如果某个ByteBuf数据没有发送完，那么该节点并不会被移除，只会调整它的readerIndex索引，下次继续发送剩余数据。
/**
* 根据写入到TCP缓冲区的字节数来移除ByteBuf。
* @param writtenBytes
*/
public void removeBytes(long writtenBytes) {
for (;;) {
// 从flushedEntry开始计算
Object msg = current();
if (!(msg instanceof ByteBuf)) {
assert writtenBytes == 0;
break;
}
final ByteBuf buf = (ByteBuf) msg;
final int readerIndex = buf.readerIndex();
final int readableBytes = buf.writerIndex() - readerIndex;
// 如果单个ByteBuf的数据 <= writtenBytes，则直接移除Entry节点
if (readableBytes <= writtenBytes) {
if (writtenBytes != 0) {
progress(readableBytes);
writtenBytes -= readableBytes;
}
remove();
} else { // readableBytes > writtenBytes
// 存在某个ByteBuf发送部分数据的情况，调整它的readerIndex，下次继续发送。
if (writtenBytes != 0) {
buf.readerIndex(readerIndex + (int) writtenBytes);
progress(writtenBytes);
}
break;
}
}
// 重置 NIO_BUFFERS
clearNioBuffers();
}
复制代码
至此，Netty的数据发送核心流程全部分析结束。
总结
为了避免每次write都将数据写入TCP缓冲区，Netty的Channel提供了两种操作：write和flush，这需要依赖一个核心类ChannelOutboundBuffer。write只是将数据暂存到缓冲区，flush才是发送数据。同时为了避免消息积压的太多导致OOM，Netty提供了高低水位线，当暂存的消息到达高水位时，Netty会将Channel设为「不可写」，同时触发回调，用户可以根据该状态判断是否需要继续写入消息。
ChannelOutboundBuffer本身是个单向链表，负责管理暂存的消息，当需要发送数据时，它还会负责将ByteBuf转换成ByteBuffer，因为JDK底层的SocketChannel只支持写入ByteBuffer。
数据发送完毕后，ChannelOutboundBuffer还要负责根据实际发送的字节数来移除Entry节点，因为存在某个ByteBuf只发送了部分数据的情况，针对这种特殊情况，ChannelOutboundBuffer不会将节点移除，而是调整它的readerIndex索引，下次继续发送剩余数据。
![[桜井宁宁]COS和泉纱雾超可爱写真福利集-一一网](https://www.proyy.com/skycj/data/images/2020-12-13/4d3cf227a85d7e79f5d6b4efb6bde3e8.jpg)
![[桜井宁宁] 爆乳奶牛少女cos写真-一一网](https://www.proyy.com/skycj/data/images/2020-12-13/d40483e126fcf567894e89c65eaca655.jpg)
