如何使 ChannelOutboundHandler.write() 方法异步？答案

【问题标题】：How to make the ChannelOutboundHandler.write() method to be asynchronous？如何使 ChannelOutboundHandler.write() 方法异步？
【发布时间】：2021-10-24 10:09:57
【问题描述】：

我的项目使用 Java Netty 框架来传输消息。该应用程序既是客户端又是服务器。当我们向远程服务器发送消息时，我们希望对这条消息进行一些处理。我在我的项目中使用ChannelOutboundHandler.write() 来达到这个目的：

public class MyOutBoundHandler extends ChannelOutboundHandlerAdapter{
    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        process((ByteBuf) msg);  // do some processing of this message
        ctx.write(msg, promise);
    }
}

我发现了一个问题，当process((ByteBuf) msg)方法抛出异常时，会导致阻塞，并且下一个方法ctx.write(msg, promise)不会被执行。那么如何使它们异步，因为我希望process((ByteBuf) msg) 不会影响将消息写入远程服务器。

【问题讨论】：

标签： java asynchronous netty

【解决方案1】：

如果 'ctx.write(msg, promise)' 不依赖于 'process((ByteBuf) msg)' 的结果，您可以将 'process((ByteBuf) msg)' 包装在一个可运行的任务中并将任务提交到线程池。

【讨论】：

【解决方案2】：

发现一个问题，当process((ByteBuf) msg)方法抛出异常会导致阻塞，下一个方法ctx.write(msg, promise)不会被执行

除非你执行阻塞代码，否则netty不会阻塞。

您正在解释的行为不是阻塞，它只是 java 中控制流的工作方式。如果抛出异常，除非显式捕获并恢复，否则不会执行任何后续代码。

理想情况下，在您的情况下，您希望在对 process() 的调用周围添加一个 try-catch 块，如果失败，则使用 promise.tryFailure() 使承诺失败

【讨论】：

【解决方案3】：

您可以在调用 process 方法之前为 ChannelPromise 添加一个 listener。

请参考以下代码：

   
        promise.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                Throwable cause = future.cause();
                if (cause != null) {

                    //process happed exception will be here and you can call ctx.write(msg) 
                    //to keep spreading forward the write event in the pipeline
                    ctx.write(msg);
                } else {
                    // when msg has been write to socket successfully, netty will notify here!!

                }
            }
        });


        process(msg);

【讨论】：