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

public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    //网络连接tcp三次握手后，就会建立和封装一个Channel(网络连接的通信管道)
    //此时这个Channel就可以实现一个激活
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Channel Active......");
        ctx.channel().writeAndFlush("test5");
    }

    @Override
    public 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);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Channel Read Complete......");
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
    private final DefaultChannelPipeline pipeline;
    ...
    @Override
    public 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;
    }
    
    @Override
    public 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;
    ...
    @Override
    public ChannelFuture writeAndFlush(Object msg) {
        return writeAndFlush(msg, newPromise());
    }
    
    @Override
    public 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();
        }
        ...
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            unsafe.write(msg, promise);
        }

        @Override
        public 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.
    @Override
    public 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. 
    @Override
    public 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.
    @Override
    public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        ctx.disconnect(promise);
    }

    //Calls ChannelHandlerContext#close(ChannelPromise) to forward to the next ChannelOutboundHandler in the ChannelPipeline.
    @Override
    public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        ctx.close(promise);
    }

    //Calls ChannelHandlerContext#deregister(ChannelPromise) to forward to the next ChannelOutboundHandler in the ChannelPipeline.
    @Override
    public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        ctx.deregister(promise);
    }

    //Calls ChannelHandlerContext#read() to forward to the next ChannelOutboundHandler in the ChannelPipeline.
    @Override
    public void read(ChannelHandlerContext ctx) throws Exception {
        ctx.read();
    }

    //Calls ChannelHandlerContext#write(Object, ChannelPromise)} to forward to the next ChannelOutboundHandler in the ChannelPipeline.
    @Override
    public 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.
    @Override
    public void flush(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }
}

//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);
    }

    @Override
    public 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 accour
    protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception;
}

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());    
    }
}

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();
        }
        ...
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            unsafe.write(msg, promise);
        }

        @Override
        public void flush(ChannelHandlerContext ctx) throws Exception {
            unsafe.flush();
        }
    }
    ...
}

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);
        ...
        @Override
        public final void write(Object msg, ChannelPromise promise) {
            //确保该方法的调用是在Reactor线程中
            assertEventLoop();
            //写缓冲区
            ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
            ...
            int size;
            try {
                //转换成堆外内存
                msg = filterOutboundMessage(msg);
                //估算出需要写入的ByteBuf的size
                size = pipeline.estimatorHandle().size(msg);
                if (size < 0) {
                    size = 0;
                }
            } catch (Throwable t) {
                safeSetFailure(promise, t);
                ReferenceCountUtil.release(msg);
                return;
            }
            //将转换成堆外内存的msg添加到写缓冲区outboundBuffer
            outboundBuffer.addMessage(msg, size, promise);
        }
        ...
    }    
    ...
}

public abstract class AbstractNioByteChannel extends AbstractNioChannel {
    ...
    @Override
    protected 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); 
    }
    ...
}

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 flushed
    private Entry flushedEntry;
    //The Entry which is the first unflushed in the linked-list structure
    private Entry unflushedEntry;
    //The Entry which represents the tail of the buffer
    private 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>() {
            @Override
            protected 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;
        }
        ...
    }
}

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();
        }
        ...
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            unsafe.write(msg, promise);
        }

        @Override
        public void flush(ChannelHandlerContext ctx) throws Exception {
            unsafe.flush();
        }
    }
    ...
}

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); 
        ...
        @Override
        public 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 {
    ...
    @Override
    protected 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);
        }
    }
    
    @Override
    protected 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 flushed
    private Entry flushedEntry;
    //The Entry which is the first unflushed in the linked-list structure
    private Entry unflushedEntry;
    //The Entry which represents the tail of the buffer
    private Entry tailEntry;
    
    private static final AtomicIntegerFieldUpdater<ChannelOutboundBuffer> UNWRITABLE_UPDATER;
    @SuppressWarnings("UnusedDeclaration")
    private volatile int unwritable;
    
    //The number of flushed entries that are not written yet
    private 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;
            }
        }
    }
    ...
}