Netty 源码解析 -- ChannelOutboundBuffer 实现与 Flush 过程

Entry tailEntry;		// 链表最后一个节点，新增的节点添加其后。
Entry unflushedEntry;	// 链表中第一个未刷新的节点
Entry flushedEntry;		// 链表中第一个已刷新但数据未写入的节点
int flushed;			// 已刷新但数据未写入的节点数

public void addMessage(Object msg, int size, ChannelPromise promise) {
	// #1
	Entry entry = Entry.newInstance(msg, size, total(msg), promise);
	if (tailEntry == null) {
		flushedEntry = null;
	} else {
		Entry tail = tailEntry;
		tail.next = entry;
	}
	tailEntry = entry;
	if (unflushedEntry == null) {
		unflushedEntry = entry;
	}

	incrementPendingOutboundBytes(entry.pendingSize, false);
}

if (ctx.channel().isActive() && ctx.channel().isWritable()) {
    ctx.writeAndFlush(responseMessage);
} else {
    ...
}

public void addFlush() {
	// #1
	Entry entry = unflushedEntry;
	if (entry != null) {
		// #2
		if (flushedEntry == null) {
			// there is no flushedEntry yet, so start with the entry
			flushedEntry = entry;
		}
		do {
			// #3
			flushed ++;
			if (!entry.promise.setUncancellable()) {
				// Was cancelled so make sure we free up memory and notify about the freed bytes
				int pending = entry.cancel();
				decrementPendingOutboundBytes(pending, false, true);
			}
			entry = entry.next;
			// #4
		} while (entry != null);

		// All flushed so reset unflushedEntry
		// #5
		unflushedEntry = null;
	}
}

public ByteBuffer[] nioBuffers(int maxCount, long maxBytes) {
	assert maxCount > 0;
	assert maxBytes > 0;
	long nioBufferSize = 0;
	int nioBufferCount = 0;
	// #1
	final InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.get();
	ByteBuffer[] nioBuffers = NIO_BUFFERS.get(threadLocalMap);
	Entry entry = flushedEntry;
	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;

			if (readableBytes > 0) {
				// #2
				if (maxBytes - readableBytes < nioBufferSize && nioBufferCount != 0) {
					break;
				}
				nioBufferSize += readableBytes;
				// #3
				int count = entry.count;
				if (count == -1) {
					//noinspection ConstantValueVariableUse
					entry.count = count = buf.nioBufferCount();
				}

				int neededSpace = min(maxCount, nioBufferCount + count);
				if (neededSpace > nioBuffers.length) {
					nioBuffers = expandNioBufferArray(nioBuffers, neededSpace, nioBufferCount);
					NIO_BUFFERS.set(threadLocalMap, nioBuffers);
				}
				// #4
				if (count == 1) {
					ByteBuffer nioBuf = entry.buf;
					if (nioBuf == null) {
						// cache ByteBuffer as it may need to create a new ByteBuffer instance if its a
						// derived buffer
						entry.buf = nioBuf = buf.internalNioBuffer(readerIndex, readableBytes);
					}
					nioBuffers[nioBufferCount++] = nioBuf;
				} else {
					...
				}
				if (nioBufferCount == maxCount) {
					break;
				}
			}
		}
		entry = entry.next;
	}
	this.nioBufferCount = nioBufferCount;
	this.nioBufferSize = nioBufferSize;

	return nioBuffers;
}

public void removeBytes(long writtenBytes) {

	for (;;) {
		// #1
		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;
		// #2
		if (readableBytes <= writtenBytes) {
			if (writtenBytes != 0) {
				progress(readableBytes);
				writtenBytes -= readableBytes;
			}
			remove();
		} else { // readableBytes > writtenBytes
			// #3
			if (writtenBytes != 0) {
				buf.readerIndex(readerIndex + (int) writtenBytes);
				progress(writtenBytes);
			}
			break;
		}
	}
	clearNioBuffers();
}

public boolean remove() {
	Entry e = flushedEntry;
	if (e == null) {
		clearNioBuffers();
		return false;
	}
	Object msg = e.msg;

	ChannelPromise promise = e.promise;
	int size = e.pendingSize;
	// #1
	removeEntry(e);

	if (!e.cancelled) {
		// only release message, notify and decrement if it was not canceled before.
		// #2
		ReferenceCountUtil.safeRelease(msg);
		safeSuccess(promise);
		decrementPendingOutboundBytes(size, false, true);
	}

	// recycle the entry
	// #3
	e.recycle();

	return true;
}

public final void flush() {
	assertEventLoop();

	ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
	if (outboundBuffer == null) {
		return;
	}
	// #1
	outboundBuffer.addFlush();
	// #2
	flush0();
}

protected void doWrite(ChannelOutboundBuffer in) throws Exception {
	SocketChannel ch = javaChannel();
	int writeSpinCount = config().getWriteSpinCount();
	do {
		// #1
		if (in.isEmpty()) {
			clearOpWrite();
			return;
		}

		// #2
		int maxBytesPerGatheringWrite = ((NioSocketChannelConfig) config).getMaxBytesPerGatheringWrite();
		ByteBuffer[] nioBuffers = in.nioBuffers(1024, maxBytesPerGatheringWrite);
		int nioBufferCnt = in.nioBufferCount();

		switch (nioBufferCnt) {
			case 0:
				// #3
				writeSpinCount -= doWrite0(in);
				break;
			case 1: {
				// #4
				ByteBuffer buffer = nioBuffers[0];
				int attemptedBytes = buffer.remaining();
				final int localWrittenBytes = ch.write(buffer);
				if (localWrittenBytes <= 0) {
					// #5
					incompleteWrite(true);
					return;
				}
				adjustMaxBytesPerGatheringWrite(attemptedBytes, localWrittenBytes, maxBytesPerGatheringWrite);
				// #6
				in.removeBytes(localWrittenBytes);
				--writeSpinCount;
				break;
			}
			default: {
				// #7
				...
			}
		}
	} while (writeSpinCount > 0);

	incompleteWrite(writeSpinCount < 0);
}

	...
    if ((readyOps & SelectionKey.OP_WRITE) != 0) {
        ch.unsafe().forceFlush();
    }