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

private void write(Object msg, boolean flush, ChannelPromise promise) {
    AbstractChannelHandlerContext next = findContextOutbound();
    final Object m = pipeline.touch(msg, next);
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        if (flush) {
            //todo 因为flush 为 true
            next.invokeWriteAndFlush(m, promise);
        } else {
            next.invokeWrite(m, promise);
        }

/**
 * @Author: Changwu
 * @Date: 2019/7/21 20:49
 */
public class MyPersonEncoder extends MessageToByteEncoder<PersonProtocol> {
    // todo write动作会传播到 MyPersonEncoder的write方法, 但是我们没有重写, 于是就执行 父类 MessageToByteEncoder的write, 我们进去看
    @Override
    protected void encode(ChannelHandlerContext ctx, PersonProtocol msg, ByteBuf out) throws Exception {
        System.out.println("MyPersonEncoder....");
        // 消息头  长度
        out.writeInt(msg.getLength());
        // 消息体
        out.writeBytes(msg.getContent());
    }
}

// todo 看他的write方法
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    ByteBuf buf = null;
    try {
        if (acceptOutboundMessage(msg)) {// todo 1  判断当前是否可以处理这个对象
            @SuppressWarnings("unchecked")
            I cast = (I) msg;
            // todo 2 内存分配
            buf = allocateBuffer(ctx, cast, preferDirect);
            try {
                // todo 3 调用本类的encode(), 这个方法就是我们自己实现的方法
                encode(ctx, cast, buf);
            } finally {
                // todo 4 释放
                ReferenceCountUtil.release(cast);
            }
            if (buf.isReadable()) {
                // todo 5. 往前传递
                ctx.write(buf, promise);
            } else {
                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) {
            // todo 释放
            buf.release();
        }
    }

@Override
    public final void write(Object msg, ChannelPromise promise) {
        assertEventLoop();
        ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
        if (outboundBuffer == null) { // todo 缓存 写进来的 buffer
            ReferenceCountUtil.release(msg);
            return;
        }
        int size;
        try {
            // todo buffer  Dirct化 , (我们查看 AbstractNioByteBuf的实现)
            msg = filterOutboundMessage(msg);
            size = pipeline.estimatorHandle().size(msg);
            if (size < 0) {
                size = 0;
            }
        } catch (Throwable t) {
            safeSetFailure(promise, t);
            ReferenceCountUtil.release(msg);
            return;
        }
        // todo 插入写队列  将 msg 插入到 outboundBuffer
        // todo outboundBuffer 这个对象是 ChannelOutBoundBuf类型的,它的作用就是起到一个容器的作用
        // todo 下面看, 是如何将 msg 添加进 ChannelOutBoundBuf中的
        outboundBuffer.addMessage(msg, size, promise);
    }

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 void addMessage(Object msg, int size, ChannelPromise promise) {
    // todo 将上面的三者封装成实体
    // todo 调用工厂方法, 创建  Entry  , 在 当前的ChannelOutboundBuffer 中每一个单位都是一个 Entry, 用它进一步包装 msg
    Entry entry = Entry.newInstance(msg, size, total(msg), promise);
    // todo 调整三个指针, 去上面查看这三个指针的定义
    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.
    // See https://github.com/netty/netty/issues/1619
    // todo 跟进这个方法
    incrementPendingOutboundBytes(entry.pendingSize, false);
}

private void incrementPendingOutboundBytes(long size, boolean invokeLater) {
    if (size == 0) {
        return;
    }
    // todo TOTAL_PENDING_SIZE_UPDATER 当前缓存中 存在的代写的 字节
    // todo 累加
    long newWriteBufferSize = TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, size);
    // todo 判断 新的将被写的 buffer的容量不能超过  getWriteBufferHighWaterMark() 默认是 64*1024  64字节
    if (newWriteBufferSize > channel.config().getWriteBufferHighWaterMark()) {
        // todo 超过64 字节,进入这个方法
        setUnwritable(invokeLater);
    }
}