netty源码阅读之编码之writeAndFlush抽象步骤

电玩女神 2022-05-09 05:38 277阅读 0赞

我们首先把对象变成字节流，最终写到socket底层的流程

![70][]

在pipeline中，从tail节点开始传播，业务流程biz这个hander处理完之后，调用writeAndFlush把对象user传递过去；然后有个encode编码器把user编码成bytebuf；最后在head里面把bytebuf变成二进制调用jdk底层的写方法写到网络里面去。

我们写了一个用户代码，

启动客户端：

public final class Server {
    
        public static void main(String[] args) throws Exception {
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
    
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new Encoder());
                        ch.pipeline().addLast(new BizHandler());
                    }
                });
    
                ChannelFuture f = b.bind(8888).sync();
    
                f.channel().closeFuture().sync();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

我们主要关注在Encoder()和BizHandler()，解码器在业务处理器之前，业务处理完之后往前传播，调用编码器把对象编码。

看BizHandler()：

public class BizHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            //...
    
            User user = new User(19, "zhangsan");
    
            ctx.channel().writeAndFlush(user);
        }
    }

关注到我们的writeAndFlush。

然后看最重要的编码器：

/**
     * ---------------------
     *|   4    |  4  |  ?   |
     * ---------------------
     *| length | age | name |
     * ---------------------
     */
    
    
    
    public class Encoder extends MessageToByteEncoder<User> {
        @Override
        protected void encode(ChannelHandlerContext ctx, User user, ByteBuf out) throws Exception {
    
            byte[] bytes = user.getName().getBytes();
            out.writeInt(4 + bytes.length);
            out.writeInt(user.getAge());
            out.writeBytes(bytes);
        }
    }

我们首先写入长度域的内容，内容就是user的age的长度（user的age的长度是int占用的字节数4）加上user的name的字节数。

后续再把user的age内容写进去，然后在写入name的内容。而写入的地方就是out，一个bytebuf，这个bytebuf后续还会使用到。

分析完这个用户代码，我们正式进去主题，writeAndFlush的步骤抽象。

1、从tail节点开始往前传播

2、逐个调用channelHandler的write方法

3、逐个调用channelHandler的flush方法

从writeAndFlush进入，

@Override
        public ChannelFuture writeAndFlush(Object msg) {
            return pipeline.writeAndFlush(msg);
        }

在pipeline里面传播。然后：

@Override
        public final ChannelFuture writeAndFlush(Object msg) {
            return tail.writeAndFlush(msg);
        }

从tail开始传播，一路看把，跳过中间一些源码到了这里：

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

判断如果是在当前线程就执行，否则就丢到任务队列里面去执行。

执行的时候如果有flush，就调用invokeWriteAndFlush，否则就只wirte就行了，我们这里以有flush的情况分析。进入invokeWriteAndFlush：

private void invokeWriteAndFlush(Object msg, ChannelPromise promise) {
            if (invokeHandler()) {
                invokeWrite0(msg, promise);
                invokeFlush0();
            } else {
                writeAndFlush(msg, promise);
            }
        }

有write也有flush，进入过invokeWrite0()：

private void invokeWrite0(Object msg, ChannelPromise promise) {
            try {
                ((ChannelOutboundHandler) handler()).write(this, msg, promise);
            } catch (Throwable t) {
                notifyOutboundHandlerException(t, promise);
            }
        }

然后进入到MessageToByteEncoder的write方法：

@Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            ByteBuf buf = null;
            try {
                if (acceptOutboundMessage(msg)) {
                    @SuppressWarnings("unchecked")
                    I cast = (I) msg;
                    buf = allocateBuffer(ctx, cast, preferDirect);
                    try {
                        encode(ctx, cast, buf);
                    } finally {
                        ReferenceCountUtil.release(cast);
                    }
    
                    if (buf.isReadable()) {
                        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) {
                    buf.release();
                }
            }
        }

这里有段：

try {
                        encode(ctx, cast, buf);
                    } finally {
                        ReferenceCountUtil.release(cast);
                    }

这个encode就是需要子类去实现的，我们上面的用户代码实现了一个Encoder。当然netty里面还有很多别的实现。

不管怎么样，netty会考虑去释放掉没使用过的资源，后面我们分析MessageToByteEncoder方法的时候会分析。

flush的也一样分析。

[70]: /images/20220509/4173a6ae07664d2d8043f29e509ee85e.png