Netty writeAndFlush() 不完整

Question

我的 tcp 客户端请求 netty 服务器，netty 服务器使用 writeAndFlush() return 393718 字节。但客户端只收到 262142 字节。我用"tcpdump -A"来抓包，也小于393718。所以我认为正确的是在 netty writeAndFlush() 函数中？

这是代码

TCP 服务器：

public static void main (String args[]) {

        int processorsNumber = Runtime.getRuntime().availableProcessors() * 3;
        ThreadFactory threadFactory = new DefaultThreadFactory("work thread pool");
        EventLoopGroup bossLoopGroup = new NioEventLoopGroup(1);
        EventLoopGroup workLoopGroup = new NioEventLoopGroup(processorsNumber, threadFactory, SelectorProvider.provider());

        try {
            ServerBootstrap sb = new ServerBootstrap();
            sb.group(bossLoopGroup , workLoopGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer<NioSocketChannel>() {
                        @Override
                        protected void initChannel(NioSocketChannel ch) throws Exception {
                            //ch.pipeline().addLast(new LineBasedFrameDecoder(1000));
                            ch.pipeline().addLast(new TcpServerHandler());
                        }
                    });

            sb.bind(new InetSocketAddress("0.0.0.0", 18881)).sync().channel().closeFuture().sync();
        } catch (InterruptedException e) {
            logger.error("netty error", e);
            e.printStackTrace();
        } finally {
            bossLoopGroup.shutdownGracefully();
            workLoopGroup.shutdownGracefully();
        }
    }

服务器处理程序：

public void channelRead(ChannelHandlerContext ctx, Object msg) {

        ByteBuf in = (ByteBuf) msg;
        String cmd = in.toString(CharsetUtil.UTF_8);
        logger.info(cmd);

        String retCode = "";
        String file = "E:\sz\app.log";
        try {
            FileReader fr = new FileReader(file);
            BufferedReader br = new BufferedReader(fr);
            String buffer = null;
            while ((buffer = br.readLine()) != null) {
                retCode += buffer;
            }
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }

        logger.info("return：" + retCode);
        byte[] result = retCode.getBytes();
        logger.info("======================="+result.length);
        ByteBuf resultBuf = Unpooled.buffer(result.length);
        resultBuf.writeBytes(result);
        ctx.writeAndFlush(resultBuf);

        ctx.close();
    }

TCP 客户端：

public static void main(String[] args) {
        StringBuilder response = new StringBuilder();
        try {
            Socket socket = new Socket();
            InetSocketAddress address = new InetSocketAddress("127.0.0.1", 18881);
            socket.connect(address, 3000);
            socket.setSoTimeout(10 * 1000);

            OutputStream clientRequest = socket.getOutputStream();
            String cmd = "cmd";
            clientRequest.write(cmd.getBytes());
            clientRequest.flush();

            InputStream clientResponse = socket.getInputStream();
            int maxLen = 1024;
            byte[] contextBytes = new byte[maxLen];
            int readLen;
            while ((readLen = clientResponse.read(contextBytes, 0, maxLen)) != -1) {
                response.append(new String(contextBytes, 0, readLen));
            }

            clientRequest.close();
            clientResponse.close();
        } catch (IOException e) {
            logger.error("tcp error", e);
            e.printStackTrace();
        }
        String result = response.toString();
        System.out.println(result);
        System.out.println(result.getBytes().length);
    }

Answer 1

写完直接关闭连接，在异步框架中，这是自找麻烦

这里发生的事情是，一条新的传入消息通过 Netty 线程到达您的 channelRead。此消息然后通过您的处理程序进行处理，最后使用 writeAndFlush 将其发回，但是，发送回数据的调用部分发生在后台，因此代码继续向前，然后它看到调用关闭连接，此时它很乐意这样做，因为你告诉它。

此时发生的是尝试发送数据的 TCP 层与接收关闭连接的调用的 TCP 层之间的竞争条件，因为在层的不同部分使用的缓冲区，它发生了因此限制为 262142 字节，但是如果连接的未知方面发生变化，此限制可能会波动。

解决这个问题的方法是等Netty说写成功了，然后关闭。由于我们不想在异步应用程序中等待，这意味着我们必须注册一个监听器，在这种情况下，幸运就在我们这边，netty 为我们提供了一个监听器，它只是为我们关闭了通道。你可以这样使用它：

ctx.writeAndFlush(resultBuf).addListener(ChannelFutureListener.CLOSE);

Answer 2

1.ctx.writeAndFlush(resultBuf) 是异步的。

write()->将bytebuf加入netty单向链表

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

incrementPendingOutboundBytes(size, false);

}

flush()-> 写入 SOCEKT 缓存。

protected int doWriteBytes(ByteBuf buf) throws Exception {
final int expectedWrittenBytes = buf.readableBytes();
return buf.readBytes(javaChannel(), expectedWrittenBytes);

}

标准话：

    // 3. writeAndFlush
    xxx.writeAndFlush().addListener(future -> {
        if (future.isDone()) {
            xxx.close();
        }
    });

Netty writeAndFlush() 不完整

Netty writeAndFlush() incomplete

java

tcp

netty

这是代码