Netty源码—8.编解码原理二

大纲

1.读数据入口

2.拆包原理

3.ByteToMessageDecoder解码步骤

4.解码器抽象的解码过程总结

5.Netty里常见的开箱即用的解码器

6.writeAndFlush()方法的大体步骤

7.MessageToByteEncoder的编码步骤

8.unsafe.write()写队列

9.unsafe.flush()刷新写队列

10.如何把对象变成字节流写到unsafe底层

6.writeAndFlush()方法的大体步骤

(1)writeAndFlush()方法的调用入口

(2)writeAndFlush()方法的执行流程

(1)writeAndFlush()方法的调用入口

入口通常是：ctx.channel().writeAndFlush()。

public class NettyServerHandler extends ChannelInboundHandlerAdapter {//网络连接tcp三次握手后，就会建立和封装一个Channel(网络连接的通信管道)//此时这个Channel就可以实现一个激活@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println("Channel Active......");ctx.channel().writeAndFlush("test5");}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {System.out.println("Channel Read: " + (String)msg);String response = "Hello World......";ByteBuf responseByteBuf = Unpooled.buffer();responseByteBuf.writeBytes(response.getBytes());ctx.channel().writeAndFlush(responseByteBuf);System.out.println("Channel Write: " + response);}@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {System.out.println("Channel Read Complete......");ctx.flush();}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}
}

(2)writeAndFlush()方法的执行流程

首先从tail结点开始往前传播。然后逐个调用ChannelHandler的write()方法，直到某个ChannelHandler不再往前传播write事件。接着逐个调用ChannelHandler的flush()方法，直到某个ChannelHandler不再往前传播flush事件。

一般而言，只要每个ChannelHandler都往下传播write事件和flush事件，那么最后都会传播到HeadContext结点的write()方法和flush()方法，然后分别执行unsafe.write()和unsafe.flush()将数据通过底层的unsafe写到JDK底层的Channel。

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {private final DefaultChannelPipeline pipeline;...@Overridepublic ChannelFuture writeAndFlush(Object msg) {return pipeline.writeAndFlush(msg);}...
}public class DefaultChannelPipeline implements ChannelPipeline {final AbstractChannelHandlerContext head;final AbstractChannelHandlerContext tail;private final Channel channel;protected DefaultChannelPipeline(Channel channel) {this.channel = ObjectUtil.checkNotNull(channel, "channel");tail = new TailContext(this);head = new HeadContext(this);head.next = tail;tail.prev = head;}@Overridepublic final ChannelFuture writeAndFlush(Object msg) {//从TailContext开始传播//但TailContext没有重写writeAndFlush()方法//所以会调用AbstractChannelHandlerContext的writeAndFlush()方法return tail.writeAndFlush(msg);}...
}abstract class AbstractChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext, ResourceLeakHint {volatile AbstractChannelHandlerContext next;volatile AbstractChannelHandlerContext prev;...@Overridepublic ChannelFuture writeAndFlush(Object msg) {return writeAndFlush(msg, newPromise());}@Overridepublic ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) {if (msg == null) throw new NullPointerException("msg");if (!validatePromise(promise, true)) {ReferenceCountUtil.release(msg);return promise;}write(msg, true, promise);return promise;}private void write(Object msg, boolean flush, ChannelPromise promise) {//反向遍历链表进行查找AbstractChannelHandlerContext next = findContextOutbound();final Object m = pipeline.touch(msg, next);EventExecutor executor = next.executor();//最终都会由Reactor线程处理Channel的数据读写if (executor.inEventLoop()) {if (flush) {//调用结点的invokeWriteAndFlush()方法next.invokeWriteAndFlush(m, promise);} else {next.invokeWrite(m, promise);}} else {AbstractWriteTask task;if (flush) {task = WriteAndFlushTask.newInstance(next, m, promise);}  else {task = WriteTask.newInstance(next, m, promise);}safeExecute(executor, task, promise, m);}}private void invokeWriteAndFlush(Object msg, ChannelPromise promise) {if (invokeHandler()) {//逐个调用ChannelHandler结点的write()方法，但前提是当前ChannelHandler可以往下传//即write()方法在最后也像ChannelOutboundHandlerAdapter那样，调用了ctx.write()往下传播invokeWrite0(msg, promise);//逐个调用ChannelHandler结点的flush()方法，但前提是当前ChannelHandler可以往下传//即flush()方法在最后也像ChannelOutboundHandlerAdapter那样，调用了ctx.flush()往下传播invokeFlush0();} else {writeAndFlush(msg, promise);}}private void invokeWrite0(Object msg, ChannelPromise promise) {try {//逐个调用，最终回到HeadContext的write()方法((ChannelOutboundHandler) handler()).write(this, msg, promise);} catch (Throwable t) {notifyOutboundHandlerException(t, promise);}}private void invokeFlush0() {try {//逐个调用，最终回到HeadContext的flush()方法((ChannelOutboundHandler) handler()).flush(this);} catch (Throwable t) {notifyHandlerException(t);}}...
}public class DefaultChannelPipeline implements ChannelPipeline {...final class HeadContext extends AbstractChannelHandlerContext implements ChannelOutboundHandler, ChannelInboundHandler {private final Unsafe unsafe;HeadContext(DefaultChannelPipeline pipeline) {super(pipeline, null, HEAD_NAME, false, true);unsafe = pipeline.channel().unsafe();setAddComplete();}...@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {unsafe.write(msg, promise);}@Overridepublic void flush(ChannelHandlerContext ctx) throws Exception {unsafe.flush();}}...
}//Skeleton implementation of a ChannelOutboundHandler. This implementation just forwards each method call via the ChannelHandlerContext.
public class ChannelOutboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelOutboundHandler {//Calls ChannelHandlerContext#bind(SocketAddress, ChannelPromise) to forward to the next ChannelOutboundHandler in the ChannelPipeline.@Overridepublic void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception {ctx.bind(localAddress, promise);}//Calls ChannelHandlerContext#connect(SocketAddress, SocketAddress, ChannelPromise) to forward to the next ChannelOutboundHandler in the ChannelPipeline. @Overridepublic void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) throws Exception {ctx.connect(remoteAddress, localAddress, promise);}//Calls ChannelHandlerContext#disconnect(ChannelPromise) to forward to the next ChannelOutboundHandler in the ChannelPipeline.@Overridepublic void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {ctx.disconnect(promise);}//Calls ChannelHandlerContext#close(ChannelPromise) to forward to the next ChannelOutboundHandler in the ChannelPipeline.@Overridepublic void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {ctx.close(promise);}//Calls ChannelHandlerContext#deregister(ChannelPromise) to forward to the next ChannelOutboundHandler in the ChannelPipeline.@Overridepublic void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {ctx.deregister(promise);}//Calls ChannelHandlerContext#read() to forward to the next ChannelOutboundHandler in the ChannelPipeline.@Overridepublic void read(ChannelHandlerContext ctx) throws Exception {ctx.read();}//Calls ChannelHandlerContext#write(Object, ChannelPromise)} to forward to the next ChannelOutboundHandler in the ChannelPipeline.@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {ctx.write(msg, promise);}//Calls ChannelHandlerContext#flush() to forward to the next ChannelOutboundHandler in the ChannelPipeline.@Overridepublic void flush(ChannelHandlerContext ctx) throws Exception {ctx.flush();}
}

7.MessageToByteEncoder的编码步骤

(1)编码的具体步骤

(2)编码步骤的总结

(3)子类实现编码的例子

(1)编码的具体步骤

步骤一：判断对象

判断当前ChannelHandler结点能否处理写入传入的Java对象。如果能处理，则往下执行，否则直接传递给下一个ChannelHandler结点进行处理。

步骤二：分配内存

给新创建的ByteBuf对象分配一块内存空间，这块内存空间将会存放由Java对象转换来的字节数据。

步骤三：调用encode

子类会实现MessageToByteEncoder的抽象方法encode()来定义自己的编码协议，子类的encode()方法会将Java对象转换来的字节数据写入ByteBuf。

步骤四：释放对象

由于传入的Java对象已经转换成ByteBuf字节流了，所以传入的Java对象已不再使用可进行释放。

步骤五：传播数据

当子类的encode()方法将数据写入了ByteBuf对象以及释放完对象之后，则会往前一个ChannelHandler结点传播该ByteBuf对象，否则往前一个ChannelHandler结点传播空对象。

步骤六：释放内存

如果出现异常或者ByteBuf没有写入数据或者ByteBuf在pipeline中已处理完，则释放分配给ByteBuf对象的内存。

//ChannelOutboundHandlerAdapter which encodes message in a stream-like fashion from one message to an ByteBuf.
//Example implementation which encodes Integers to a ByteBuf.
//public class IntegerEncoder extends MessageToByteEncoder<Integer> {
//    @code @Override
//    public void encode(ChannelHandlerContext ctx, Integer msg, ByteBuf out) throws Exception {
//        out.writeInt(msg);
//    }
//}
public abstract class MessageToByteEncoder<I> extends ChannelOutboundHandlerAdapter {private final TypeParameterMatcher matcher;private final boolean preferDirect;protected MessageToByteEncoder() {this(true);}protected MessageToByteEncoder(Class<? extends I> outboundMessageType) {this(outboundMessageType, true);}//Create a new instance which will try to detect the types to match out of the type parameter of the class.//@param preferDirect，true if a direct ByteBuf should be tried to be used as target for the encoded messages. //If false is used it will allocate a heap ByteBuf, which is backed by an byte array.protected MessageToByteEncoder(boolean preferDirect) {matcher = TypeParameterMatcher.find(this, MessageToByteEncoder.class, "I");this.preferDirect = preferDirect;}//Create a new instance//@param outboundMessageType，The tpye of messages to match//@param preferDirect，true if a direct ByteBuf should be tried to be used as target for the encoded messages. //If false is used it will allocate a heap ByteBuf, which is backed by an byte array.protected MessageToByteEncoder(Class<? extends I> outboundMessageType, boolean preferDirect) {matcher = TypeParameterMatcher.get(outboundMessageType);this.preferDirect = preferDirect;}//Returns true if the given message should be handled. //If false it will be passed to the next ChannelOutboundHandler in the ChannelPipeline.public boolean acceptOutboundMessage(Object msg) throws Exception {return matcher.match(msg);}@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {ByteBuf buf = null;try {//步骤一：判断当前ChannelHandler能否处理写入的消息if (acceptOutboundMessage(msg)) {@SuppressWarnings("unchecked")I cast = (I) msg;//强制转换//步骤二：给ByteBuf对象分配内存buf = allocateBuffer(ctx, cast, preferDirect);try {//步骤三：调用子类实现的encode()方法encode(ctx, cast, buf);} finally {//步骤四：释放对象//既然自定义的Java对象msg已经转换为ByteBuf对象了，那么该对象已经没有用，需要释放掉了//注意：当传入的msg的类型是ByteBuf类型时，则不需要释放ReferenceCountUtil.release(cast);}//步骤五：如果buf中写入了数据，就把buf传到下一个ChannelHandler结点if (buf.isReadable()) {ctx.write(buf, promise);} else {//步骤六：如果buf中没有写入数据，则释放buf，并将一个空数据传到下一个ChannelHandler结点buf.release();ctx.write(Unpooled.EMPTY_BUFFER, promise);}buf = null;} else {ctx.write(msg, promise);}} catch (EncoderException e) {throw e;} catch (Throwable e) {throw new EncoderException(e);} finally {if (buf != null) {buf.release();//当buf在pipeline中处理完了，需要进行释放}}}//Allocate a ByteBuf which will be used as argument of #encode(ChannelHandlerContext, I, ByteBuf).//Sub-classes may override this method to returna ByteBuf with a perfect matching initialCapacity.protected ByteBuf allocateBuffer(ChannelHandlerContext ctx, @SuppressWarnings("unused") I msg, boolean preferDirect) throws Exception { if (preferDirect) {return ctx.alloc().ioBuffer();} else {return ctx.alloc().heapBuffer();}}//Encode a message into a ByteBuf. //This method will be called for each written message that can be handled by this encoder.//@param ctx，the ChannelHandlerContext which this MessageToByteEncoder belongs to//@param msg，the message to encode//@param out，the ByteBuf into which the encoded message will be written//@throws Exception，is thrown if an error accourprotected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception;
}

(2)编码步骤的总结

在MessageToByteEncoder的编码过程中，首先会判断当前ChannelHandler能否处理传入的Java对象，如果能处理就对新创建的ByteBuf对象分配一块内存空间。然后由子类的encode()方法实现具体的编码协议，并且把编码后的数据存放到分配给ByteBuf对象的内存空间中。最后把ByteBuf对象往前一个ChannelHandler结点进行传播。

如果在编码的过程中出现异常，那么就把已申请出来的、分配给ByteBuf对象的内存空间进行释放。

如果传入的Java对象就是一个ByteBuf对象，那么Netty在自定义编码结束后，会自动帮忙释放该对象，不需要在子类中对该对象进行释放。

(3)子类实现编码的例子

下面的Encoder便实现了将自定义的Response对象转换为字节流并写到Socket底层的效果。

public class Encoder extends MessageToByteEncoder<Response> {protected void encode(ChannelHandlerContext ctx, Response response, ByteBuf out) throws Exception {out.writeByte(response.getVersion());out.writeInt(4+ response.getData().length);out.writeBytes(response.getData());    }
}

8.unsafe.write()将数据添加到写缓冲区

(1)unsafe.write()的入口

(2)unsafe.write()的主要逻辑

(3)写缓冲区(写队列)的数据结构

(1)unsafe.write()的入口

不管是ctx.channel().write()还是ctx.write()，最终都会来到pipeline中的head结点。

public class DefaultChannelPipeline implements ChannelPipeline {...final class HeadContext extends AbstractChannelHandlerContext implements ChannelOutboundHandler, ChannelInboundHandler {private final Unsafe unsafe;HeadContext(DefaultChannelPipeline pipeline) {super(pipeline, null, HEAD_NAME, false, true);unsafe = pipeline.channel().unsafe();setAddComplete();}...@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {unsafe.write(msg, promise);}@Overridepublic void flush(ChannelHandlerContext ctx) throws Exception {unsafe.flush();}}...
}

(2)unsafe.write()的主要逻辑

unsafe.write()方法将数据添加到写缓冲区(写队列)的主要逻辑如下。

一.Direct化ByteBuf对象

如果传进来的ByteBuf对象不是堆外内存，那么就直接转换成堆外内存，并且估算出其大小。

二.添加到写缓冲区

转换成堆外内存的ByteBuf对象首先会被封装成一个Entry对象，然后再将该Entry对象添加到写缓冲区，其中会通过几个指针来标识写缓冲区的状态。

三.设置写状态

如果内存不足，那么是不可以一直往写缓冲区里添加ByteBuf对象的。如果写缓冲区已经大于默认的64KB的大小，则会通过自旋 + CAS设置当前Channel为不可写状态。

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {private final DefaultChannelPipeline pipeline;...protected abstract class AbstractUnsafe implements Unsafe {//写缓冲区(写队列)private volatile ChannelOutboundBuffer outboundBuffer = new ChannelOutboundBuffer(AbstractChannel.this);...@Overridepublic final void write(Object msg, ChannelPromise promise) {//确保该方法的调用是在Reactor线程中assertEventLoop();//写缓冲区ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;...int size;try {//转换成堆外内存msg = filterOutboundMessage(msg);//估算出需要写入的ByteBuf的sizesize = pipeline.estimatorHandle().size(msg);if (size < 0) {size = 0;}} catch (Throwable t) {safeSetFailure(promise, t);ReferenceCountUtil.release(msg);return;}//将转换成堆外内存的msg添加到写缓冲区outboundBufferoutboundBuffer.addMessage(msg, size, promise);}...}    ...
}public abstract class AbstractNioByteChannel extends AbstractNioChannel {...@Overrideprotected final Object filterOutboundMessage(Object msg) {if (msg instanceof ByteBuf) {ByteBuf buf = (ByteBuf) msg;if (buf.isDirect()) {return msg;}return newDirectBuffer(buf);}if (msg instanceof FileRegion) {return msg;}throw new UnsupportedOperationException("unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES); }...
}

(3)写缓冲区(写队列)的数据结构

ChannelOutboundBuffer里的数据结构是一个单向链表，单向链表的每个结点都是一个Entry对象。在一个Entry对象中会包含着待写出的ByteBuf对象及消息回调promise。flushedEntry指针表示第一个被写入Socket缓冲区的结点，unflushedEntry指针表示第一个未被写入Socket缓冲区的结点，tailEntry指针表示ChannelOutboundBuffer缓冲区的最后一个结点。

初次调用ChannelOutboundBuffer的addMessage()方法后，flushedEntry指针指向NULL，unflushedEntry指针和tailEntry指针都指向新添加的结点。调用多次ChannelOutboundBuffer的addMessage()方法后，如果flushedEntry指针一直指向NULL，则表示现在还没有结点的ByteBuf对象写出到Socket缓冲区。如果unflushedEntry指针之后有n个结点，则表示当前还有n个结点的ByteBuf对象还没写出到Socket缓冲区。

public final class ChannelOutboundBuffer {private final Channel channel;//Entry(flushedEntry) --> ... Entry(unflushedEntry) --> ... Entry(tailEntry)//The Entry that is the first in the linked-list structure that was flushedprivate Entry flushedEntry;//The Entry which is the first unflushed in the linked-list structureprivate Entry unflushedEntry;//The Entry which represents the tail of the bufferprivate Entry tailEntry;...ChannelOutboundBuffer(AbstractChannel channel) {this.channel = channel;}//Add given message to this ChannelOutboundBuffer. //The given {@link ChannelPromise} will be notified once the message was written.public void addMessage(Object msg, int size, ChannelPromise promise) {Entry entry = Entry.newInstance(msg, size, total(msg), promise);if (tailEntry == null) {flushedEntry = null;tailEntry = entry;} else {Entry tail = tailEntry;tail.next = entry;tailEntry = entry;}if (unflushedEntry == null) {unflushedEntry = entry;}//increment pending bytes after adding message to the unflushed arrays.incrementPendingOutboundBytes(size, false);}static final class Entry {private static final Recycler<Entry> RECYCLER = new Recycler<Entry>() {@Overrideprotected Entry newObject(Handle handle) {return new Entry(handle);}};private final Handle<Entry> handle;Entry next;Object msg;ByteBuffer[] bufs;ByteBuffer buf;ChannelPromise promise;long progress;long total;int pendingSize;int count = -1;boolean cancelled;private Entry(Handle<Entry> handle) {this.handle = handle;}static Entry newInstance(Object msg, int size, long total, ChannelPromise promise) {Entry entry = RECYCLER.get();entry.msg = msg;entry.pendingSize = size;entry.total = total;entry.promise = promise;return entry;}...}
}

9.unsafe.flush()刷新写缓冲区的数据

(1)unsafe.flush()的入口

(2)unsafe.flush()的主要逻辑

(1)unsafe.flush()的入口

不管是ctx.channel().flush()还是ctx.flush()，最终都会来到pipeline中的head结点。

public class DefaultChannelPipeline implements ChannelPipeline {...final class HeadContext extends AbstractChannelHandlerContext implements ChannelOutboundHandler, ChannelInboundHandler {private final Unsafe unsafe;HeadContext(DefaultChannelPipeline pipeline) {super(pipeline, null, HEAD_NAME, false, true);unsafe = pipeline.channel().unsafe();setAddComplete();}...@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {unsafe.write(msg, promise);}@Overridepublic void flush(ChannelHandlerContext ctx) throws Exception {unsafe.flush();}}...
}

(2)unsafe.flush()的主要逻辑

步骤一：设置flushedEntry指针指向unflushedEntry指针所指向的Entry结点，并统计需要刷新的Entry结点的数量。

步骤二：遍历写缓冲区的Entry结点把对应的ByteBuf对象写到Socket，然后移除Entry结点。如果写缓冲区大小已经小于32KB，则通过自旋 + CAS设置Channel为可写状态。

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {private final DefaultChannelPipeline pipeline;...protected abstract class AbstractUnsafe implements Unsafe {private volatile ChannelOutboundBuffer outboundBuffer = new ChannelOutboundBuffer(AbstractChannel.this); ...@Overridepublic final void flush() {assertEventLoop();ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;if (outboundBuffer == null) {return;}//步骤一outboundBuffer.addFlush();//步骤二flush0();}protected void flush0() {final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;...doWrite(outboundBuffer);...}}//Flush the content of the given buffer to the remote peer.protected abstract void doWrite(ChannelOutboundBuffer in) throws Exception;
}public abstract class AbstractNioByteChannel extends AbstractNioChannel {...@Overrideprotected void doWrite(ChannelOutboundBuffer in) throws Exception {//默认自旋16次，以提高内存使用率和写的吞吐量int writeSpinCount = config().getWriteSpinCount();do {Object msg = in.current();if (msg == null) {//重新注册，不关注OP_WRITE事件clearOpWrite();return;}writeSpinCount -= doWriteInternal(in, msg);} while(writeSpinCount > 0);incompleteWrite(setOpWrite);}private int doWriteInternal(ChannelOutboundBuffer in, Object msg) {...ByteBuf buf = (ByteBuf) msg;if (!buf.isReadable()) {//从写缓冲区(写队列)中移除结点in.remove();return 0;}//把ByteBuf对象写到Socket里final int localFlushedAmount = doWriteBytes(buf);if (localFlushedAmount > 0) {in.progress(localFlushedAmount);if (!buf.isReadable()) {//从写缓冲区(写队列)中移除结点in.remove();}return 1;}...}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. if (!key.isValid()) {return;}final int interestOps = key.interestOps();if ((interestOps & SelectionKey.OP_WRITE) != 0) {key.interestOps(interestOps & ~SelectionKey.OP_WRITE);}}@Overrideprotected int doWriteBytes(ByteBuf buf) throws Exception {final int expectedWrittenBytes = buf.readableBytes();return buf.readBytes(javaChannel(), expectedWrittenBytes);}...
}public final class ChannelOutboundBuffer {private final Channel channel;//Entry(flushedEntry) --> ... Entry(unflushedEntry) --> ... Entry(tailEntry)//The Entry that is the first in the linked-list structure that was flushedprivate Entry flushedEntry;//The Entry which is the first unflushed in the linked-list structureprivate Entry unflushedEntry;//The Entry which represents the tail of the bufferprivate Entry tailEntry;private static final AtomicIntegerFieldUpdater<ChannelOutboundBuffer> UNWRITABLE_UPDATER;@SuppressWarnings("UnusedDeclaration")private volatile int unwritable;//The number of flushed entries that are not written yetprivate int flushed;...//设置flushedEntry指针指向unflushedEntry指针所指向的Entry结点，//并统计需要刷新的Entry结点的数量public void addFlush() {Entry entry = unflushedEntry;if (entry != null) {if (flushedEntry == null) {flushedEntry = entry;}do {flushed ++;//所要flush的结点数entry = entry.next;} while (entry != null);unflushedEntry = null;}}public boolean remove() {//获取当前正在被flush的结点Entry e = flushedEntry;Object msg = e.msg;//获取该结点的回调对象ChannelPromise promise = e.promise;int size = e.pendingSize;//从写缓冲队列中移除结点removeEntry(e);if (!e.cancelled) {ReferenceCountUtil.safeRelease(msg);safeSuccess(promise);//如果写缓冲区大小小于32KB，就通过自旋+CAS设置Channel状态为可写decrementPendingOutboundBytes(size, false, true);}//回收实体e.recycle();return true;}private void removeEntry(Entry e) {if (-- flushed == 0) {flushedEntry = null;if (e == tailEntry) {tailEntry = null;unflushedEntry = null;}} else {flushedEntry = e.next;}}//Return the current message to write or null if nothing was flushed before and so is ready to be written.public Object current() {Entry entry = flushedEntry;if (entry == null) {return null;}return entry.msg;}//Notify the ChannelPromise of the current message about writing progress.public void progress(long amount) {Entry e = flushedEntry;assert e != null;ChannelPromise p = e.promise;if (p instanceof ChannelProgressivePromise) {long progress = e.progress + amount;e.progress = progress;((ChannelProgressivePromise) p).tryProgress(progress, e.total);}}private void decrementPendingOutboundBytes(long size, boolean invokeLater, boolean notifyWritability) {if (size == 0) {return;}long newWriteBufferSize = TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, -size);if (notifyWritability && newWriteBufferSize < channel.config().getWriteBufferLowWaterMark()) {setWritable(invokeLater);}}private void setWritable(boolean invokeLater) {for (;;) {final int oldValue = unwritable;final int newValue = oldValue & ~1;if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {if (oldValue != 0 && newValue == 0) {fireChannelWritabilityChanged(invokeLater);}break;}}}...
}

10.如何把对象变成字节流写到unsafe底层

当调用ctx.channel().writeAndFlush(user)将自定义的User对象沿着整个Pipeline进行传播时：

首先会调用tail结点的write()方法开始往前传播，传播到一个继承自MessageToByteEncoder的结点。该结点会实现MessageToByteEncoder的encode()方法来把自定义的User对象转换成一个ByteBuf对象。转换的过程首先会由MessageToByteEncoder分配一个ByteBuf对象，然后再调用其子类实现的抽象方法encode()将User对象填充到ByteBuf对象中。填充完之后继续调用write()方法把该ByteBuf对象往前进行传播，默认下最终会传播到head结点。

其中head结点的write()方法会通过底层的unsafe进行如下处理：把当前的ByteBuf对象添加到unsafe维护的一个写缓冲区里，同时计算写缓冲区大小是否超过64KB。如果写缓冲区大小超过了64KB，则设置当前Channel不可写。完成write()方法的传播后，head结点的unsafe对象维护的写缓冲区便对应着一个ByteBuf队列，它是一个单向链表。

然后会调用tail结点的flush()方法开始往前传播，默认下最终会传播到head结点。head结点在接收到flush事件时会通过底层的unsafe进行如下处理：首先进行指针调整，然后通过循环遍历从写缓冲区里把ByteBuf对象取出来。每拿出一个ByteBuf对象都会把它转化为JDK底层可以接受的ByteBuffer对象，最终通过JDK的Channel把该ByteBuffer对象写出去。每写完一个ByteBuffer对象都会把写缓冲区里的当前ByteBuf所在的Entry结点进行删除，并且判断如果当前写缓冲区里的大小已经小于32KB就通过自旋 + CAS重新设置Channel为可写。