有关高水位问题，数据不能刷出 netty 缓存

[lichao306153702]

//Server code
`public class EchoServer {

public static void main(String[] args) throws InterruptedException {
    ServerBootstrap serverBootstrap = new ServerBootstrap();
    NioEventLoopGroup group = new NioEventLoopGroup();
    NioEventLoopGroup work = new NioEventLoopGroup();
    serverBootstrap.group(group, work).channel(NioServerSocketChannel.class)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    socketChannel.pipeline().addLast(new EchoServerHandle());
                }
            })
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(50 * 1024, 100 * 1024));
    ChannelFuture f = serverBootstrap.bind("192.168.5.126", 50099).addListener((ChannelFutureListener) future -> {
        if (future.isSuccess()) {
            System.out.println("bing success" + future.channel().localAddress());
        }
    });
    ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.PARANOID);
    f.channel().closeFuture().sync();
}

}
`

//server handle
`public class EchoServerHandle extends ChannelInboundHandlerAdapter {

public static final AtomicInteger num = new AtomicInteger(0);

private static int size = 1024 * 100;  // 100kb


@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    ByteBuf buf = (ByteBuf) msg;
    byte[] bytes = new byte[buf.readableBytes()];
    buf.readBytes(bytes);
    buf.release();
    System.out.println("receive：" + new String(bytes));
}

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    StringBuilder content = new StringBuilder("1");
    for (int i = 0; i < size - 1; i++) {  //100kb
        content.append("*");
    }
    while (ctx.channel().isActive()) {
        if (ctx.channel().isWritable()) {
            num.incrementAndGet();
            System.out.println("发送：" + num.get());
            ByteBuf byteBuf = PooledByteBufAllocator.DEFAULT.directBuffer();
            byteBuf.writeInt(content.length());  //  4
            byteBuf.writeBytes(content.toString().getBytes());  //
            ctx.channel().writeAndFlush(byteBuf);
        } else {
            System.out.println("高水位");
            int size = ctx.channel().unsafe().outboundBuffer().size();
            System.out.println(size);
            ((AbstractNioChannel) ctx.channel()).unsafe().forceFlush();
            int size1 = ctx.channel().unsafe().outboundBuffer().size();
            System.out.println(size1);
            TimeUnit.SECONDS.sleep(1);
        }
    }

}

}`

// client code

`public class EchoClient {

public static void main(String[] args) throws InterruptedException {
    NioEventLoopGroup group = new NioEventLoopGroup();
    Bootstrap bootstrap = new Bootstrap();
    bootstrap.group(group)
            .channel(NioSocketChannel.class)
            .remoteAddress("192.168.5.126", 50099)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    socketChannel.pipeline().addLast(new MyDecoder(1024 * 1024 * 20, 0, 4));
                    socketChannel.pipeline().addLast(new EchoClientHandle());
                }
            })
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Integer.MAX_VALUE)
            .option(ChannelOption.SO_REUSEADDR, true)
            .option(ChannelOption.AUTO_READ, true)
            //需要设置这个
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .option(ChannelOption.SO_KEEPALIVE, true);

    ChannelFuture channelFuture = bootstrap.connect().sync();
    ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.PARANOID);
    channelFuture.channel().closeFuture().sync();
}

}
`

// client handle
`
public class EchoClientHandle extends ChannelInboundHandlerAdapter {
public static final AtomicInteger num = new AtomicInteger(0);

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    Message message = (Message) msg;
    TimeUnit.MILLISECONDS.sleep(20);
    ByteBuf head = message.getHead();
    byte[] content = new byte[head.readableBytes()];
    head.readBytes(content);
    String s = new String(content);
    System.out.println("get：" + s.substring(4, 10) + ",num=" + num.incrementAndGet());
    head.release();
}

}
`
//MyDecoder code

`public class MyDecoder extends LengthFieldBasedFrameDecoder {

public MyDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength) {
    super(maxFrameLength, lengthFieldOffset, lengthFieldLength);
}

@Override
protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
    ByteBuf decode = (ByteBuf) super.decode(ctx, in);
    if (decode == null) {
        return null;
    }
    ByteBuf head = decode.slice(0, 10);
    ByteBuf body = decode.slice(10, decode.readableBytes()-10);
    Message message = new Message(head, body);
    return message;
}

}
`

/ / description
The server starts, the client starts, and when the data is sent to the high water level, the data is not sent
You need to call this code, force a refresh,
((AbstractNioChannel) ctx.channel()).unsafe().forceFlush();
Am I using it incorrectly?

[needmorecode]

达到高水位后缓冲满了，就是没法write了呀。你的意思是直接调channel.flush()也发不出去？

[lichao306153702]

对的，，可以通过我粘贴的代码，测试一下，

[normanmaurer]

Sorry but I can only help if you use English

[lichao306153702]

Okay, so my problem is that when I'm on the server and I keep feeding data into the channel, and it hits a high water level, the channel stays at a high water level, and I can't send messages, and the client can't receive messages,

[normanmaurer]

You are "blocking" the EventLoop with your while (ctx.channel().isActive()). Don't do that as it will basically disallow any other IO operation to run.

[ma-meng]

yes, you are right.

[KevinBlandy]

我之前看过讲述类似问题的文章，大致思想是这样的

你应该监听，channelReadComplete 事件，读取完客户端的请求数据后，判断一下通道是否可写（isWritable）。如果通道不可写，也就是缓冲区满了。那你可以设置客户端（channel）setAutoRead(false)。不再读取客户端的输入，直到服务端的缓存队列变为可写状态。

监听 channelWritabilityChanged 事件，在这个事件里面判断，如果通道可写了，就设置客户端 channel.config().setAutoRead(true);。重新开始读取客户端的输入。

通俗的理解就是，服务端的缓存满了。就不再处理客户端数据，（在recv buf满了后，客户端也没法发送数据）直到服务端缓存队列可用了，在重新开始处理客户端的请求。