深入理解 Netty编码流程及WriteAndFlush()的实现 （二）

// todo 最终传递到 这里
@Override
public final void flush() {
    assertEventLoop();
    ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
    if (outboundBuffer == null) {
        return;
    }
    // todo 添加刷新标志, 设置写状态
    outboundBuffer.addFlush();
    // todo  遍历buffer队列, 过滤byteBuf
    flush0();
}

* todo 给 ChannelOutboundBuffer 添加缓存, 这意味着, 原来添加进 ChannelOutboundBuffer 中的所有 Entry, 全部会被标记为 flushed 过
 */
public void addFlush() {
// todo 默认让 entry 指向了 unflushedEntry ==> 其实链表中的最左边的 未被使用过的 entry
// todo
Entry entry = unflushedEntry;
if (entry != null) {
    if (flushedEntry == null) {
        // there is no flushedEntry yet, so start with the entry
        flushedEntry = entry;
    }
    do {
        flushed ++;
        if (!entry.promise.setUncancellable()) {
            // Was cancelled so make sure we free up memory and notify about the freed bytes
            int pending = entry.cancel();
            // todo 跟进这个方法
            decrementPendingOutboundBytes(pending, false, true);
        }
        entry = entry.next;
    } while (entry != null);
    // All flushed so reset unflushedEntry
    unflushedEntry = null;
}
}

if (flushedEntry == null) {
    // there is no flushedEntry yet, so start with the entry
    flushedEntry = entry;
}

private void decrementPendingOutboundBytes(long size, boolean invokeLater, boolean notifyWritability) {
    if (size == 0) {
        return;
    }
    // todo 每次 减去 -size
    long newWriteBufferSize = TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, -size);
    // todo   默认 getWriteBufferLowWaterMark() -32kb
    // todo   newWriteBufferSize<32 就把不可写状态改为可写状态
    if (notifyWritability && newWriteBufferSize < channel.config().getWriteBufferLowWaterMark()) {
        setWritable(invokeLater);
    }
}

boolean setOpWrite = false;
        // todo 整体是无限循环, 过滤ByteBuf
for (;;) {
    // todo 获取第一个 flushedEntity, 这个entity中 有我们需要的 byteBuf
    Object msg = in.current();
    if (msg == null) {
        // Wrote all messages.
        clearOpWrite();
        // Directly return here so incompleteWrite(...) is not called.
        return;
    }
    if (msg instanceof ByteBuf) {
        // todo 第三部分,jdk底层, 进行自旋的写
        ByteBuf buf = (ByteBuf) msg;
        int readableBytes = buf.readableBytes();
        if (readableBytes == 0) {
            // todo 当前的 ByteBuf 中,没有可写的, 直接remove掉
            in.remove();
            continue;
        }
        boolean done = false;
        long flushedAmount = 0;
        if (writeSpinCount == -1) {
            // todo 获取自旋锁, netty使用它进行
            writeSpinCount = config().getWriteSpinCount();
        }
        // todo 这个for循环是在自旋尝试往 jdk底层的 ByteBuf写入数据
        for (int i = writeSpinCount - 1; i >= 0; i --) {
            // todo  把 对应的 buf , 写到socket中
            // todo localFlushedAmount就是 本次 往jdk底层的 ByteBuffer 中写入了多少字节
            int localFlushedAmount = doWriteBytes(buf);
            if (localFlushedAmount == 0) {
                setOpWrite = true;
                break;
            }
            // todo 累加一共写了多少字节
            flushedAmount += localFlushedAmount;
            // todo 如果buf中的数据全部写完了, 设置完成的状态, 退出循环
            if (!buf.isReadable()) {
                done = true;
                break;
            }
        }
        in.progress(flushedAmount);
        // todo 自旋结束,写完了  done = true
        if (done) {
            // todo 跟进去
            in.remove();
        } else {
            // Break the loop and so incompleteWrite(...) is called.
            break;
        }
    ....

public Object current() {
        Entry entry = flushedEntry;
        if (entry == null) {
            return null;
        }
        return entry.msg;
    }

// todo
@Override
protected int doWriteBytes(ByteBuf buf) throws Exception {
    final int expectedWrittenBytes = buf.readableBytes();
    // todo 将字节数据, 写入到 java 原生的 channel中
    return buf.readBytes(javaChannel(), expectedWrittenBytes);
}

// todo
    @Override
    public int readBytes(GatheringByteChannel out, int length) throws IOException {
        checkReadableBytes(length);
        //todo  关键的就是 getBytes()  跟进去
        int readBytes = getBytes(readerIndex, out, length, true);
        readerIndex += readBytes;
        return readBytes;
    }
    跟进getBytes(){
        index = idx(index);
        // todo 将netty 的 ByteBuf 塞进 jdk的    ByteBuffer tmpBuf;
        tmpBuf.clear().position(index).limit(index + length);
        // todo 调用jdk的write()方法
        return out.write(tmpBuf);
    }

private void removeEntry(Entry e) {
        if (-- flushed == 0) { // todo 如果是最后一个节点, 把所有的指针全部设为 null
            // processed everything
            flushedEntry = null;
            if (e == tailEntry) {
                tailEntry = null;
                unflushedEntry = null;
            }
        } else { //todo 如果 不是最后一个节点, 把当前节点,移动到最后的 节点
            flushedEntry = e.next;
        }
    }