从零开始实现简易版Netty(四) MyNetty 高效的数据写出实现

宗和玉 · 2025-7-18 19:03:11

从零开始实现简易版Netty(四) MyNetty 高效的数据写出实现

1. MyNetty 数据写出处理优化

在上一篇博客中，lab3版本的MyNetty对事件循环中的IO读事件处理做了一定的优化，解决了之前版本无法进行大数据量读取的问题。
按照计划，本篇博客中，lab4版本的MyNetty需要实现高效的数据写出。由于本文属于系列博客，读者需要对之前的博客内容有所了解才能更好地理解本文内容。

lab1版本博客：从零开始实现简易版Netty(一) MyNetty Reactor模式
lab2版本博客：从零开始实现简易版Netty(二) MyNetty pipeline流水线
lab3版本博客：从零开始实现简易版Netty(三) MyNetty 高效的数据读取实现

目前版本的实现中，MyNetty允许用户在自定义的ChannelHandler中使用ChannelHandlerContext的write方法向远端写出消息。
write操作被视为一个出站事件会一直从后往前传播到head节点，最终在pipeline流水线的head节点中通过jdk的socketChannel.write方法将消息写出。

public class MyChannelPipelineHeadContext extends MyAbstractChannelHandlerContext implements MyChannelEventHandler {
// 已省略无关逻辑
@Override
public void write(MyChannelHandlerContext ctx, Object msg) throws Exception {
SocketChannel socketChannel = (SocketChannel) ctx.getPipeline().getChannel().getJavaChannel();
if(msg instanceof ByteBuffer){
socketChannel.write((ByteBuffer) msg);
}else{
// msg走到head节点的时候，必须是ByteBuffer类型
throw new Error();
}
}
}

复制代码

面对大量的待读取消息，Netty针对性的实现了启发式算法动态调节buffer容器大小、限制单channel读事件中读取消息的次数以避免其它channel饥饿等机制。
同样的，如果需要短时间内向外写出大量的消息时，目前版本MyNetty中简单的socketChannel.write来实现写出功能是远远不够的，极端场景下同样存在很多问题。

操作系统向外发送数据的缓冲区是有限的，当网络拥塞、系统负载过高等场景时，通过socketChannel.write写出的数据并不能总是完全成功的写出，而是可能部分甚至全部都无法写出。此时，未成功写出的数据需要由应用程序自己维护好，等待缓冲区空闲后进行重试。
write方法的执行是异步的，事件循环线程a发出的消息可能实际上是由线程b来真正的写出，而线程b又不总是能一次成功的写出消息。
MyNetty中线程a目前无法感知到write方法发送的消息是否被成功写出，同时也无法在缓冲区负载较高时，主动限制自己写出消息的速率而在源头实现限流。
写出操作与读取操作一样，都是需要占用cpu的，即使write方法写出的消息总是能一次性的成功发出，也不能让当前channel无限制的写出数据，而导致同处一个事件循环中的其它channel饥饿。

因此在本篇博客中，lab4版本的MyNetty需要参考netty对写出操作进行改进，优化大量数据写出时的各种问题。
2. MyNetty 高效的数据写出实现源码解析

在解析MyNetty的源码之前，先分析一下netty是如何解决上述问题的。

针对操作系统缓冲区有限，可能在网络拥塞等场景无法立即写出数据的问题，Netty内部为每一个Channel都设置了一个名为ChannelOutboundBuffer的应用层缓冲区。
通过write方法写出的消息数据，会先暂存在该缓冲区中。而只有通过flush/writeAndFlush方法触发flush操作时，才会实际的往channel的对端写出。
当所暂存的消息无法完全写出时，消息也不会丢失，而是会一直待在缓冲区内等待缓冲区空闲后重试写出(具体原理在后面解析)。
针对write方法异步操作而无法令用户感知结果的问题，netty的write方法返回了一个类似Future的ChannelPromise对象，用户可以通过注册回调方法来感知当前所写出消息的结果。
针对用户无法感知当前系统负载情况而无法主动限流的问题，netty引入了高水位、低水位的概念。
当channel对应的ChannelOutboundBuffer中待写出消息总大小高于某个阈值时，被认为处于高水位，负载较高，不能继续写入了；而待写出消息的总大小低于某个阈值时，被认为处于低水位，负载较低，可以继续写入了。
用户可以通过channel的isWritable方法主动的查询当前的负载状态(返回false代表负载较高，不可写)，也可以监听channelHandler的channelWritabilityChanged获取isWritable状态的转换结果。
基于isWritable标识，用户能够在负载较高时主动的调节自己发送消息的速率，避免待发送消息越写越多而最终导致OOM。
针对单个channel内待写出消息过多而可能导致同一EventLoop内其它channel饥饿的问题，与避免读事件饥饿的机制一样，netty同样限制了一次完整的写出操作内可以写出的消息次数。
在写出次数超过阈值时，则立即终止当前channel的写出，转而处理其它事件/任务；当前channel内未写完的消息留待后续的事件循环中再接着处理。

MyNetty ChannelOutboundBuffer实现源码

public class MyChannelOutboundBuffer {
private static final Logger logger = LoggerFactory.getLogger(MyChannelOutboundBuffer.class);
private static final int DEFAULT_LOW_WATER_MARK = 32 * 1024;
private static final int DEFAULT_HIGH_WATER_MARK = 64 * 1024;
private final MyNioChannel channel;
private MyChannelOutBoundBufferEntry flushedEntry;
private MyChannelOutBoundBufferEntry unFlushedEntry;
private MyChannelOutBoundBufferEntry tailEntry;
public int num;
/**
* 简单一点，这里直接用boolean类型
* netty中用整型，是为了同时表达多个index下的unWritable语义(setUserDefinedWritability)
* */
private volatile boolean unWritable;
/**
* 当前outBoundBuffer
* 目前只支持持有channel的IO线程更新，所以不设置为volatile
* */
private long totalPendingSize;
/**
* 一次写出操作的nioBuffer集合的大小总和
* */
private long nioBufferSize;
// The number of flushed entries that are not written yet
private int flushed;
private static final ThreadLocal<ByteBuffer[]> NIO_BUFFERS = ThreadLocal.withInitial(() -> new ByteBuffer[1024]);
public MyChannelOutboundBuffer(MyNioChannel channel) {
this.channel = channel;
}
public void addMessage(ByteBuffer msg, int size, CompletableFuture<MyNioChannel> completableFuture) {
// 每个msg对应一个链表中的entry对象
MyChannelOutBoundBufferEntry entry = MyChannelOutBoundBufferEntry.newInstance(msg, size, completableFuture);
if (tailEntry == null) {
// 当前队列为空
flushedEntry = null;
} else {
// 当前待flush的队列不为空，新节点追加到tail上
MyChannelOutBoundBufferEntry tail = tailEntry;
tail.next = entry;
}
tailEntry = entry;
// 当前的unFlushed队列为空，当前消息作为头结点
if (unFlushedEntry == null) {
unFlushedEntry = entry;
}
// 加入新的msg后，当前outBoundBuffer所占用的总字节数自增
incrementPendingOutboundBytes(entry.pendingSize);
}
public void addFlush() {
// There is no need to process all entries if there was already a flush before and no new messages
// where added in the meantime.
//
// See https://github.com/netty/netty/issues/2577
MyChannelOutBoundBufferEntry entry = unFlushedEntry;
if(entry == null) {
// 触发flush操作，如果此时unFlushedEntry为null，说明没有需要flush的消息了，直接返回(针对重复的flush操作进行性能优化)
return;
}
if (flushedEntry == null) {
// there is no flushedEntry yet, so start with the entry
// 当前需要flush的队列(flushedEntry为head)为空，调转指针，将unFlushed队列中的消息全部转为待flushedEntry(相当于重新取了一批数据来写出)
flushedEntry = entry;
}
// 允许用户在flush之前取消掉此次写入,将待flush的节点最后检查一遍
do {
flushed ++;
if (entry.completableFuture.isCancelled()) {
// 设置cancel标志位，标识为不需要去写入
entry.cancel();
// 如果用户自己cancel了这次write操作，直接自减totalPendingSize即可
decrementPendingOutboundBytes(entry.pendingSize);
}
entry = entry.next;
} while (entry != null);
// flush操作完毕后，unFlushed队列为空
unFlushedEntry = null;
}
/**
* 按照写出的字节数，将已经写完的byteBuffer清除掉。
*
* 由于缓冲区可能受限，socketChannel.write实际没有完整的写出一个byteBuffer，这种情况ByteBuffer的remaining实际上是减少了(写出了多少减多少)
* */
public void removeBytes(long totalWrittenBytes) {
for (;;) {
MyChannelOutBoundBufferEntry currentEntry = currentEntry();
if(currentEntry == null){
// 已flushed的节点都遍历完成了
return;
}
final int readableBytes = currentEntry.msg.remaining();
if (readableBytes == 0) {
// 总共写出的bytes自减掉对应消息的大小
totalWrittenBytes -= currentEntry.msgSize;
// 完整的写出了一个byteBuffer，将其移除掉
remove();
} else {
// readableBytes > writtenBytes
// 发现一个未写完的ByteBuffer，不能移除，退出本次处理。等待下一次继续写出
return;
}
}
}
public boolean remove() {
MyChannelOutBoundBufferEntry entry = flushedEntry;
if (entry == null) {
return false;
}
CompletableFuture<MyNioChannel> completableFuture = entry.completableFuture;
int size = entry.pendingSize;
removeEntry(entry);
if (!entry.cancelled) {
// 写入操作flush成功，通知future
try {
completableFuture.complete(this.channel);
}catch (Throwable ex) {
logger.error("MyChannelOutboundBuffer notify write complete error! channel={}",this.channel,ex);
}
decrementPendingOutboundBytes(size);
}
return true;
}
/**
* 参考netty的ChannelOutboundBuffer的nioByteBuffers方法，因为没有ByteBuf到ByteBuffer的转换，所以简单不少
* */
public List<ByteBuffer> nioByteBuffers(int maxCount, int maxBytes) {
long totalNioBufferSize = 0;
// 简单起见，需要处理的byteBuffer列表直接new出来，暂不考虑优化
List<ByteBuffer> needWriteByteBufferList = new ArrayList<>();
MyChannelOutBoundBufferEntry entry = flushedEntry;
// 遍历队列中所有已经flush的节点
while (isFlushedEntry(entry)){
// 只处理未cancel的节点
if(!entry.cancelled) {
// 和netty不同，这里直接msg就是jdk的ByteBuffer，直接操作msg即可，不需要转换
int readableBytes = entry.msg.remaining();
// 只处理可读的消息，空msg忽略掉
if (readableBytes > 0) {
// 判断一下是否需要将当前的msg进行写出，如果超出了maxBytes就留到下一次再处理
// 判断!byteBufferList.isEmpty的目的是避免一个超大的msg直接超过了maxBytes
// 如果是这种极端情况即byteBufferList.isEmpty，且 readableBytes > maxBytes,那也要尝试着进行写出
// 让底层的操作系统去尽可能的写入，不一定要一次写完，下次再进来就能继续写(readableBytes会变小)
if (maxBytes < totalNioBufferSize + readableBytes && !needWriteByteBufferList.isEmpty()) {
break;
}
// 总共要写出的bufferSize自增
totalNioBufferSize += readableBytes;
// 当前msg加入待写出的list中
needWriteByteBufferList.add(entry.msg);
if (needWriteByteBufferList.size() >= maxCount) {
// 限制一下一次写出最大的msg数量
break;
}
}
}
// 遍历下一个节点
entry = entry.next;
}
this.nioBufferSize = totalNioBufferSize;
return needWriteByteBufferList;
}
public long getNioBufferSize() {
return nioBufferSize;
}
public boolean isWritable() {
return !unWritable;
}
public boolean isEmpty() {
return flushed == 0;
}
private void removeEntry(MyChannelOutBoundBufferEntry e) {
// 已flush队列头节点出队，队列长度自减1，
flushed--;
if (flushed == 0) {
// 当前已flush的队列里的消息已经全部写出完毕，将整个队列整理一下
// 首先已flush队列清空
flushedEntry = null;
if (e == tailEntry) {
// 如果当前写出的是队列里的最后一个entry，说明所有的消息都写完了，整个队列清空
tailEntry = null;
unFlushedEntry = null;
}
} else {
// 当前已flush的队列里还有剩余的消息待写出，已flush队列的头部出队，队列头部指向下一个待写出节点
flushedEntry = e.next;
}
}
private boolean isFlushedEntry(MyChannelOutBoundBufferEntry e) {
return e != null && e != unFlushedEntry;
}
public MyChannelOutBoundBufferEntry currentEntry() {
MyChannelOutBoundBufferEntry entry = flushedEntry;
if (entry == null) {
return null;
}
return entry;
}
private void incrementPendingOutboundBytes(long size) {
if (size == 0) {
return;
}
this.totalPendingSize += size;
if (totalPendingSize > DEFAULT_HIGH_WATER_MARK) {
// 超过了所配置的高水位线，标识设置为不可写
this.unWritable = true;
}
}
private void decrementPendingOutboundBytes(long size) {
if (size == 0) {
return;
}
this.totalPendingSize -= size;
if (totalPendingSize < DEFAULT_LOW_WATER_MARK) {
// 低于了所配置的低水位线，标识设置为可写
this.unWritable = false;
}
}
}

复制代码

class MyChannelOutBoundBufferEntry {
// Assuming a 64-bit JVM:
// - 16 bytes object header
// - 6 reference fields
// - 2 long fields
// - 2 int fields
// - 1 boolean field
// - padding
// netty中ChannelOutboundBuffer的Entry对象属性较多，16 + 6*8 + 2*8 + 2*4 + 1 = 89
// 往大了计算，未开启指针压缩时，64位机器按照8的倍数向上取整，算出填充默认需要96字节(netty中可通过系统参数(io.netty.transport.outboundBufferEntrySizeOverhead)动态配置)
// MyNetty做了简化，暂时没那么属性，但这里就不改了，只是多浪费了一些空间
// 详细的计算方式可参考大佬的博客：https://www.cnblogs.com/binlovetech/p/16453634.html
private static final int DEFAULT_CHANNEL_OUTBOUND_BUFFER_ENTRY_OVERHEAD = 96;
MyChannelOutBoundBufferEntry next;
ByteBuffer msg;
CompletableFuture<MyNioChannel> completableFuture;
int msgSize;
int pendingSize;
boolean cancelled;
static MyChannelOutBoundBufferEntry newInstance(ByteBuffer msg, int msgSize, CompletableFuture<MyNioChannel> completableFuture) {
// 简单起见，暂时不使用对象池，直接new
MyChannelOutBoundBufferEntry entry = new MyChannelOutBoundBufferEntry();
entry.msg = msg;
entry.msgSize = msgSize;
// entry实际的大小 = 消息体的大小 + 对象头以及各个属性值占用的大小
entry.pendingSize = msgSize + DEFAULT_CHANNEL_OUTBOUND_BUFFER_ENTRY_OVERHEAD;
entry.completableFuture = completableFuture;
return entry;
}
void cancel() {
if (!cancelled) {
cancelled = true;
}
}
}

复制代码

EventLoop中对write事件的处理逻辑

public abstract class MyNioChannel {
// 已省略无关代码
public void doWrite(Object msg, boolean doFlush, CompletableFuture<MyNioChannel> completableFuture) throws IOException {
if(!(msg instanceof ByteBuffer)){
// 约定好，msg走到head节点的时候，只支持ByteBuffer类型
throw new Error();
}
ByteBuffer byteBufferMsg = (ByteBuffer)msg;
MyChannelOutboundBuffer myChannelOutboundBuffer = this.myChannelOutboundBuffer;
// netty在存入outBoundBuffer时使用的是堆外内存缓冲，避免积压过多的数据造成堆内存移除
// 这里简单起见先不考虑这方面的性能优化，重点关注ChannelOutboundBuffer本身的功能实现
myChannelOutboundBuffer.addMessage(byteBufferMsg,byteBufferMsg.limit(),completableFuture);
if(doFlush){
myChannelOutboundBuffer.addFlush();
// 进行实际的写出操作
flush0();
}
}
public void flush0(){
if(myChannelOutboundBuffer.isEmpty()){
// 没有需要flush的消息，直接返回
return;
}
// netty针对当前channel的状态做了很多判断(isActive、isOpen)，避免往一个不可用的channel里写入数据，简单起见先不考虑这些场景
try {
doWrite(myChannelOutboundBuffer);
}catch (Exception e){
logger.error("flush0 doWrite error! close channel={}",this,e);
// 写出时有异常时，关闭channel
this.channelPipeline.close();
}
}
public boolean isWritable() {
MyChannelOutboundBuffer buf = this.myChannelOutboundBuffer;
return buf != null && buf.isWritable();
}
protected abstract void doWrite(MyChannelOutboundBuffer channelOutboundBuffer) throws Exception;
protected final void setOpWrite() {
final SelectionKey key = selectionKey;
// Check first if the key is still valid as it may be canceled as part of the deregistration
// from the EventLoop
// See https://github.com/netty/netty/issues/2104
if (!key.isValid()) {
return;
}
final int interestOps = key.interestOps();
if ((interestOps & SelectionKey.OP_WRITE) == 0) {
key.interestOps(interestOps | SelectionKey.OP_WRITE);
}
}
protected final void clearOpWrite() {
final SelectionKey key = selectionKey;
// Check first if the key is still valid as it may be canceled as part of the deregistration
// from the EventLoop
// See https://github.com/netty/netty/issues/2104
if (!key.isValid()) {
return;
}
final int interestOps = key.interestOps();
if ((interestOps & SelectionKey.OP_WRITE) != 0) {
// 去掉对于op_write事件的监听
key.interestOps(interestOps & ~SelectionKey.OP_WRITE);
}
}
protected final void incompleteWrite(boolean setOpWrite) {
// Did not write completely.
if (setOpWrite) {
setOpWrite();
} else {
// It is possible that we have set the write OP, woken up by NIO because the socket is writable, and then
// use our write quantum. In this case we no longer want to set the write OP because the socket is still
// writable (as far as we know). We will find out next time we attempt to write if the socket is writable
// and set the write OP if necessary.
clearOpWrite();
// Schedule flush again later so other tasks can be picked up in the meantime
// Calling flush0 directly to ensure we not try to flush messages that were added via write(...) in the
// meantime.
myNioEventLoop.execute(this::flush0);
}
}
}

复制代码

public class MyNioSocketChannel extends MyNioChannel{
// 已省略无关代码
/**
* 一次聚合写出的最大字节数
* */
private int maxBytesPerGatheringWrite = 1024 * 1024 * 1024;
public static final int MAX_BYTES_PER_GATHERING_WRITE_ATTEMPTED_LOW_THRESHOLD = 4096;
@Override
protected void doWrite(MyChannelOutboundBuffer myChannelOutboundBuffer) throws Exception {
// 默认一次写出16次
int writeSpinCount = 16;
do {
if (myChannelOutboundBuffer.isEmpty()) {
// 当前积压的待flush消息已经写完了，清理掉注册的write监听
// All written so clear OP_WRITE
clearOpWrite();
// Directly return here so incompleteWrite(...) is not called.
return;
}
// 计算出当前这一次写出的bytebuffer的数量
List<ByteBuffer> needWriteByteBufferList = myChannelOutboundBuffer.nioByteBuffers(1024,maxBytesPerGatheringWrite);
// 相比netty里用数组，List会多一次数组copy，但这样简单一点，不用考虑缓存or动态扩容的问题，暂不优化
ByteBuffer[] byteBuffers = needWriteByteBufferList.toArray(new ByteBuffer[0]);
SocketChannel socketChannel = this.getSocketChannel();
// 调用jdk channel的write方法一次性写入byteBuffer集合
final long localWrittenBytes = socketChannel.write(byteBuffers,0, needWriteByteBufferList.size());
logger.info("localWrittenBytes={},attemptedBytes={},needWriteByteBufferList.size={}"
, localWrittenBytes, myChannelOutboundBuffer.getNioBufferSize(),needWriteByteBufferList.size());
if (localWrittenBytes <= 0) {
// 返回值localWrittenBytes小于等于0，说明当前Socket缓冲区写满了，不能再写入了。注册一个OP_WRITE事件(setOpWrite=true)，
// 当channel所在的NIO循环中监听到当前channel的OP_WRITE事件时，就说明缓冲区又可写了，在对应逻辑里继续执行写入操作
incompleteWrite(true);
// 既然写不下了就直接返回，不需要继续尝试了
return;
}
long attemptedBytes = myChannelOutboundBuffer.getNioBufferSize();
// 基于本次写出的情况，动态的调整一次写出的最大字节数maxBytesPerGatheringWrite
adjustMaxBytesPerGatheringWrite((int) attemptedBytes, (int) localWrittenBytes, maxBytesPerGatheringWrite);
// 按照实际写出的字节数进行计算，将写出完毕的ByteBuffer从channelOutboundBuffer中移除掉
myChannelOutboundBuffer.removeBytes(localWrittenBytes);
// 每次写入一次消息，writeSpinCount自减
writeSpinCount--;
} while (writeSpinCount > 0);
// 自然的退出了循环，说明已经正确的写完了writeSpinCount指定条数的消息，但channelOutboundBuffer还不为空(如果写完了会提前return)
// incompleteWrite内部提交一个flush0的任务，等待到下一次事件循环中再捞出来处理，保证不同channel间读写的公平性
incompleteWrite(false);
}
private void adjustMaxBytesPerGatheringWrite(int attempted, int written, int oldMaxBytesPerGatheringWrite) {
// By default we track the SO_SNDBUF when ever it is explicitly set. However some OSes may dynamically change
// SO_SNDBUF (and other characteristics that determine how much data can be written at once) so we should try
// make a best effort to adjust as OS behavior changes.
// 默认情况下，我们会追踪明确设置SO_SNDBUF的地方。(setSendBufferSize等)
// 然而，一些操作系统可能会动态更改SO_SNDBUF（以及其他决定一次可以写入多少数据的特性），
// 因此我们应该尽力根据操作系统的行为变化进行调整。
if (attempted == written) {
// 本次操作写出的数据能够完全写出，说明操作系统当前还有余力
if (attempted << 1 > oldMaxBytesPerGatheringWrite) { // 左移1位，大于的判断可以保证maxBytesPerGatheringWrite不会溢出为负数
// 进一步判断，发现实际写出的数据比指定的maxBytesPerGatheringWrite要大一倍以上
// 则扩大maxBytesPerGatheringWrite的值，在后续尽可能多的写出数据
// 通常在maxBytesPerGatheringWrite较小，而某一个消息很大的场景下会出现(nioBuffers方法)
this.maxBytesPerGatheringWrite = attempted << 1;
}
} else if (attempted > MAX_BYTES_PER_GATHERING_WRITE_ATTEMPTED_LOW_THRESHOLD && written < attempted >>> 1) {
// 如果因为操作系统底层缓冲区不够的原因导致实际写出的数据量(written)比需要写出的数据量(attempted)低了一倍以上,可能是比较拥塞或者其它原因(配置或动态变化)
// 将一次写出的最大字节数缩小为原来的一半，下次尝试少发送一些消息，以提高性能
this.maxBytesPerGatheringWrite = attempted >>> 1;
}
}
}

复制代码

public class MyNioEventLoop implements Executor {
// 已省略无关代码
private void processSelectedKeys() throws IOException {
// processSelectedKeysPlain
Iterator<SelectionKey> selectionKeyItr = unwrappedSelector.selectedKeys().iterator();
while (selectionKeyItr.hasNext()) {
SelectionKey key = selectionKeyItr.next();
logger.info("process SelectionKey={}",key.readyOps());
try {
// 拿出来后，要把集合中已经获取到的事件移除掉，避免重复的处理
selectionKeyItr.remove();
if (key.isConnectable()) {
// 处理客户端连接建立相关事件
processConnectEvent(key);
}
if (key.isAcceptable()) {
// 处理服务端accept事件（接受到来自客户端的连接请求）
processAcceptEvent(key);
}
if (key.isReadable()) {
// 处理read事件
processReadEvent(key);
}
if(key.isWritable()){
// 处理OP_WRITE事件（setOpWrite中注册的）
processWriteEvent(key);
}
}catch (Throwable e){
logger.error("server event loop process an selectionKey error!",e);
// 处理io事件有异常，取消掉监听的key，并且尝试把channel也关闭掉
key.cancel();
if(key.channel() != null){
logger.error("has error, close channel={} ",key.channel());
key.channel().close();
}
}
}
}
private void processWriteEvent(SelectionKey key) throws IOException {
// 目前所有的attachment都是MyNioChannel
MyNioSocketChannel myNioChannel = (MyNioSocketChannel) key.attachment();
// 执行flush0方法
myNioChannel.flush0();
}
}

复制代码

MyChannelOutboundBuffer中维护了两个单向链表结构，一个是unFlushedEntry作为头结点的待刷新链表，一个是flushedEntry作为头结点的已刷新待写出的链表。通过write方法想要写出的消息都会被包装成entry节点被放在待刷新链表中。
其中，所接受的消息必须是ByteBuffer类型。因此应用必须在写出操作的链路上，最终将自定义消息对象通过编码处理器统一转换为ByteBuffer。
而当flush=true且已刷新待写出链表为空时，会通过一次巧妙的引用切换，将待刷新链表中的所有节点都转换到已刷新链表中。同时flush0方法会遍历已刷新链表，将其中的数据按照顺序通过socketChannel.write向对端进行实际的写出操作。
如果一个节点中对应所有消息被完整的写出时，会将该节点从已刷新链表中摘除。如果因为拥塞或消息体很大无法一次完全写出，则会继续留在已刷新链表内。

netty中定义了低水位和高水位两个阈值。每个消息在放入缓冲区时，会计算对应链表节点所占用总大小，当缓冲区中全部链表节点所占用的总大小超过了高水位阈值时，则unWritable变为true。用户感知到该变化时，则应该避免继续写入而堆积过量消息导致OOM。而当消息随着正常的写出，被从已刷新链表中被不断删除时，所占用的总空间则会慢慢减少，当减少到低于低水位阈值时，unWritable变为false，代表有空闲，可以继续写入消息了。
与处理读事件一样，netty在写出时也通过启发式的算法来动态调整下一次批量写出消息的数据量。既能避免拥塞，又能用尽可能少的次数将同样大小的消息体发出。

退出消息写出逻辑的三种方式

当前flush操作后，已刷新链表中的消息都正常的写出成功了(myChannelOutboundBuffer.isEmpty=true)。
实际写出时，socketChannel.write方法返回等于0，说明缓冲区拥塞已不可写，继续尝试短时间内已不太可能成功，通过向EventLoop注册可写事件的监听后结束此次处理。在下一次事件循环时，processWriteEvent中会再次执行flush0进行重试，继续消息的写出。
缓冲区一直可写，但是待写出消息数据量过大，超过了单次可允许批量写出的次数后依然无法全部写出，则向EventLoop注册一个task任务后结束处理。
再后续事件循环中，该任务会被捞取出来继续重试。通过这种机制，使得同一EventLoop中的其它IO事件和task任务能够有机会被处理，避免其它channel饥饿。

总结

在lab4中，MyNetty参考netty优化了写出操作的处理逻辑，支持写出大数据量的消息，并且能够让用户感知当前channel写出消息堆积的情况和监听写出操作的结果。在保留了Netty关于写操作处理最核心逻辑的基础上，省略了许多旁路逻辑以减轻读者的理解负担，比如各种阈值参数的动态配置(MyNetty里都是直接写死)、未支持channelWritabilityChanged事件等等。
在这里十分推荐大佬bin的技术小屋的博客一文搞懂 Netty 发送数据全流程 | 你想知道的细节全在这里，里面对于netty写出功能分析的非常完善，相信在理解了MyNetty的lab1-lab4的内容后，读者能更好的理解大佬博客中所涉及到的netty中各模块整体的交互逻辑。

博客中展示的完整代码在我的github上：https://github.com/1399852153/MyNetty (release/lab4_efficient_write 分支)，内容如有错误，还请多多指教。

来源：豆瓜网用户自行投稿发布，如果侵权，请联系站长删除

账号		自动登录	找回密码
密码			立即注册

从零开始实现简易版Netty(四) MyNetty 高效的数据写出实现

相关帖子

浏览过的版块

签约作者

从零开始实现简易版Netty(四) MyNetty 高效的数据写出实现

相关帖子

相关推荐

浏览过的版块

签约作者