java - 为什么netty不能连续发送消息？

Question

我编写了一个服务器，用于在连接后向所有客户端发送大量消息。

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    while (true){
       content = arrayblockqueue.poll()
       ctx.writeAndFlush(content+"\r\n");
    }
}

发送数千条消息后， channel 不再发送消息。通过调试，发现网络拥塞时会调用AbstractNioByteChannel.incompleteWrite，并且selectionKey会添加SelectionKey.OP_WRITE。设置OP_WRITE后，AbstractNioUnsafe.isFlushPending()将返回true，因此flush()确实无法完成。如何让netty恢复这种情况呢？或者我以错误的方式使用netty？

Answer 1

您的处理程序方法是直接从 I/O 线程调用的。在处理程序方法返回之前，调用处理程序方法的 I/O 线程无法执行任何 I/O，这就是为什么您看不到任何内容写入的原因。

从您的代码来看，您想要的是从阻塞队列中获取消息并将其写入 channel 。您可以直接写入 channel ，而不是使用阻塞队列。 Netty 中几乎所有操作都是线程安全的。例如:

public static void main(String[] args) throws Exception {
    ...
    Channel ch = ...;
    for (int i = 0; i < 1000000; i ++) {
        ch.writeAndFlush(String.valueOf(i) + "\r\n");
    }
    ...
}

// And your handler doesn't need an arrayblockingqueue.

但是，上面的代码很可能会让Netty的事件队列无限增长，导致OutOfMemoryError。为了防止写入请求无限排队，您必须使用 writeAndFlush() 操作返回的 future。

for (int i = 0; i < 1000000; i ++) {
    ChannelFuture f = ch.writeAndFlush(String.valueOf(i) + "\r\n");
    if ((i + 1) % 100 == 0) {
        // Wait until the write request is actually finished
        // so that the event queue becomes empty.
        f.sync();  
    }
}