Netty源码学习7——netty是如何发送数据的

一丶Write事件的产生和传播

在业务逻辑处理完毕后，需要调用write 或者 writeAndFlush方法

ChannelHandlerContext#write or writeAndFlush方法会从当前 ChannelHandler 开始在 pipeline 中向前传播 write 事件直到 HeadContext。

ChannelHandlerContext.channel()#write or writeAndFlush 方法则会从 pipeline 的尾结点 TailContext 开始在 pipeline 中向前传播 write 事件直到 HeadContext 。

write 方法并不是真将数据写到socket缓存区，而是写道Netty的ChannelOutBoundBuffer中，调用flush方法才会真正调用JDK SockectChannel将数据写入。

如下是pipeline#writeAndFlush，可以看到直接调用TailContext#writeAndFlush进行处理

关键源码如下：

private void write(Object msg, boolean flush, ChannelPromise promise) {
   // 省略 部分
	
    //flush 表示是否需要flush，调用writeAndFlush的时候为true
    // 找到下一个ChannelHandlerContext
    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()) {
        // 需要flush 那么调用invokeWriteAndFlush
        if (flush) {
            next.invokeWriteAndFlush(m, promise);
        } else {
            next.invokeWrite(m, promise);
        }
    } else {
        // 在eventLoop 中 那么提交一个WriteTask到eventLoop中
        final WriteTask task = WriteTask.newInstance(next, m, promise, flush);
        if (!safeExecute(executor, task, promise, m, !flush)) {
            task.cancel();
        }
    }
}

可以看到TailContext会通过 findContextOutbound 方法在当前 ChannelHandler 的前边找到 ChannelOutboundHandler 类型并且覆盖实现 write 回调方法的 ChannelHandler 作为下一个要执行的对象。

然后如果当前执行的线程就是EventLoop线程，那么直接调用，反之提交一个异步任务，从而保证执行write的一定是 reactor 线程——保证线程安全性

如下是next.invokeWriteAndFlush的源码

最终事件会传播到HeadContext进行处理（如果中间的ChannelHandler不截胡的话）

二丶write 源码解析

write 事件最终会由HeadContext进行处理

可以看到HeadContext#write其实就是使用Channel的Unsafe#write，其主要逻辑如下

ChannelOutboundBuffer#addMessage

ChannelOutboundBuffer 是 Netty 内部使用的一个数据结构，它用于存储待发送的出站数据。在 Netty 的网络框架中，当需要写数据到网络时，数据并不会立即被发送出去，而是首先被放入一个出站缓冲区中，即 ChannelOutboundBuffer。这个缓冲区负责管理和存储所有待写入通道的数据。

批量发送优化： ChannelOutboundBuffer 允许 Netty 批量地发送数据，而不是每次写操作都立即进行网络发送。这样可以减少系统调用次数，提高网络效率。

流量控制：它有助于实现流量控制，防止数据发送过快，导致接收方处理不过来。

缓冲区管理：可以有效地管理内存，当数据被写入网络后，及时释放相应的内存。

异步处理： Netty 是异步事件驱动的框架，使用 ChannelOutboundBuffer 可以将数据发送的异步化，提升处理性能

下面是向ChannelOutboundBuffer写入messge的源码

可与看到ChannelOutboundBuffer会将msg和promise包装为Entry，然后改变tailEntry，flushedEntry，unflushedEntry指针的指向

然后incrementPendingOutboundBytes将记录下待写出数据size，如果大于高水位还会触发channelWritabilityChanged事件

channelWritabilityChanged会在pipeline上传播，并触发ChannelInboundHandler#channelWritabilityChanged，我们可以实现此方法调用flush将数据写出

三丶flush源码解析

上面看了write将待发送的数据缓存到ChannelOutboundBuffer中，正真将数据写到SocketChannel中的是flush方法

1.addFlush

此方法只是负责更改flushedEntry 和 unflushedEntry 指针指向

将 flushedEntry 指针指向 unflushedEntry 指针表示的第一个未被 flush 的 Entry 节点。并将 unflushedEntry 指针置为空，准备开始 flush 发送数据流程。

这样在 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;
        }
    }

2.flush0

可以看到如果注册了write到selector上，那么不会进行flush，

如下是NioSockectChannel发送数据的源码

@Override
    protected void doWrite(ChannelOutboundBuffer in) throws Exception {
        //获取jdk nio底层socketChannel
        SocketChannel ch = javaChannel();
        //最大写入次数 默认为16 ，因为EventLoop可能单线程处理多Channel，需要雨露均沾
        int writeSpinCount = config().getWriteSpinCount();
        do {
            if (in.isEmpty()) {
                // 如果全部数据已经写完 则移除OP_WRITE事件并直接退出writeLoop
                clearOpWrite();             
                return;
            }

            // 获取单次发送最大字节数
            int maxBytesPerGatheringWrite = ((NioSocketChannelConfig) config).getMaxBytesPerGatheringWrite();
            //Netty的DirectBuffer底层就是JDK的DirectByteBuffer
            // 将ChannelOutboundBuffer中缓存的DirectBuffer转换成JDK的ByteBuffer，
            ByteBuffer[] nioBuffers = in.nioBuffers(1024, maxBytesPerGatheringWrite);
            // ChannelOutboundBuffer中总共的DirectBuffer数
            int nioBufferCnt = in.nioBufferCount();

            switch (nioBufferCnt) {
                    // 真正进行发送
               //java.nio.channels.SocketChannel#write(java.nio.ByteBuffer)进行写回
            }
        } while (writeSpinCount > 0);
        // 处理Socket可写但已经写满16次还没写完的情况
     incompleteWrite(writeSpinCount < 0);
    }

可以看到

如果数据全部写完了，会调用clearOpWrite清除当前 Channel 在 Reactor 上注册的 OP_WRITE 事件

这意味着，不需要再监听write来触发flush了

写入的过程会写入多次，并控制自旋次数，做到雨露均沾

如上是写入的过程

如果ByteBuffer个数为0，说明发送的是FileRegion 类型，case 0的分支主要就是用于处理网络文件传输的情况

case1 和 default则调用jdk SocketChannel#write进行数据发送，如果写入的数据小于等于0，说明当前Socket发送缓冲区满了写不进去了，则注册OP_WRITE事件，等待Socket发送缓冲区可写时再写

触发Write后，再Sockect写缓冲区可写后，会触发对应事件，即可再NioEventLoop中进行处理，如下图中会直接调用forceFlush

完成发送会调用adjustMaxBytesPerGatheringWrite进行调整

两个分支分别表示

期望写入和真正写入的相等，说明数据能全部写入到 Socket 的写缓冲区中了，那么下次 write loop 就应该尝试去写入更多的数据。

本次写入的数量x2>maxBytesPerGatheringWrite 说明要写的数据很多,那么更新为本次 write loop 两倍的写入量大小

如果本次写入的数据还不及尝试写入数据的一半，说明Socket写缓冲区容量不多了，尝试缩容为一半

处理

protected final void incompleteWrite(boolean setOpWrite) {
    
        if (setOpWrite) {
            //socket缓冲区已满写不进去的情况 注册write事件
            setOpWrite();
        } else {
            //处理socket缓冲区依然可写，但是写了16次还没写完,提交flushTask异步写
            clearOpWrite();
            eventLoop().execute(flushTask);

        }