出站 ChannelHandler 的包罗万象的异常处理

Question

在 Netty 中，您有入站和出站处理程序的概念。一个包罗万象的入站异常处理程序只需在管道的末端（尾部）添加一个通道处理程序并实现exceptionCaught 覆盖即可实现。沿入站管道发生的异常将沿处理程序传播，直到遇到最后一个，如果沿途未处理。

传出处理程序并没有完全相反的情况。相反（根据 Netty in Action，第 94 页），您需要将侦听器添加到 channel 的 Future 或将侦听器添加到 Promise 传递到您的 write 方法的 @987654326 @。

由于我不确定在哪里插入前者，我想我会选择后者，所以我做了以下ChannelOutboundHandler：

}

/**
 * Catch and log errors happening in the outgoing direction
 *
 * @see <p>p94 in "Netty In Action"</p>
 */
private ChannelOutboundHandlerAdapter createOutgoingErrorHandler() {
    return new ChannelOutboundHandlerAdapter() {
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
            logger.info("howdy! (never gets this far)");

            final ChannelFutureListener channelFutureListener = future -> {
                if (!future.isSuccess()) {
                    future.cause().printStackTrace();
                    // ctx.writeAndFlush(serverErrorJSON("an error!"));
                    future.channel().writeAndFlush(serverErrorJSON("an error!"));
                    future.channel().close();
                }
            };
            promise.addListener(channelFutureListener);
            ctx.write(msg, promise);
        }
    };

这是添加到管道的头部：

@Override
public void addHandlersToPipeline(final ChannelPipeline pipeline) {
    pipeline.addLast(
            createOutgoingErrorHandler(),
            new HttpLoggerHandler(), // an error in this `write` should go "up"
            authHandlerFactory.get(),
            // etc

问题是如果我在HttpLoggerHandler.write() 中抛出运行时异常，我的错误处理程序的write 方法永远不会被调用。

我将如何进行这项工作？任何传出处理程序中的错误都应该“冒泡”到附加到头部的错误。

需要注意的重要一点是，我不仅想关闭频道，我还想将错误消息写回客户端（从serverErrorJSON('...') 可以看出。在我尝试改组的过程中处理程序（也尝试来自this answer的东西），我已经激活了监听器，但我无法写任何东西。如果我在监听器中使用ctx.write()，看起来好像我进入了一个循环，同时使用@ 987654334@ 什么都没做。

Answer 1

基本上你所做的是正确的......唯一不正确的是处理程序的顺序。您的ChannelOutboundHandlerAdapter mast 被放置在管道中“作为最后一个出站处理程序”。这意味着它应该是这样的：

pipeline.addLast(
        new HttpLoggerHandler(),
        createOutgoingErrorHandler(),
        authHandlerFactory.get());

原因是出站事件从管道的尾部流向头部，而入站事件从头部流向尾部。

Answer 2

似乎没有一个通用的概念，即用于传出处理程序的包罗万象的异常处理程序，无论在哪里都会捕获错误。这意味着，除非您注册了一个侦听器来捕获某个错误，否则运行时错误可能会导致该错误被“吞下”，让您为为什么没有返回任何内容而摸不着头脑。

也就是说，让一个处理程序/侦听器总是在错误的情况下执行（因为它需要非常通用）可能没有意义，但它确实使日志记录错误比需要的要复杂一些。

在写完a bunch of learning tests（我建议检查一下！）之后，我得到了这些见解，它们基本上是我的 JUnit 测试的名称（经过一些正则表达式操作）：

• 监听器可以在父级写入完成后写入通道
• 写入侦听器可以从管道中移除侦听器并在错误写入时写入
• 如果传递相同的 Promise，则成功调用所有侦听器
• 尾部附近的错误处理程序无法从头部附近的处理程序中捕获错误
• netty 不会调用下一个处理程序写入运行时异常
• netty 在正常写入时调用一次写入侦听器
• netty 在写入错误时调用一次写入侦听器
• netty 使用其写入的消息调用下一个处理程序写入
• promises 可用于监听下一个处理程序的成功或失败
• 如果 Promise 被传递，Promises 可用于监听非即时处理程序的结果
• 如果传递了新的 Promise，则 Promise 不能用于监听非即时处理程序的结果
• 如果未传递承诺，则承诺不能用于侦听非即时处理程序的结果
• 如果未传递承诺，则仅在错误时调用添加到最终写入的侦听器
• 如果未传递承诺，则仅在成功时调用添加到最终写入的侦听器
• 从尾部调用写入侦听器

根据问题中的示例，这种见解意味着，如果在尾部附近出现错误并且 authHandler 没有传递承诺，那么头部附近的错误处理程序将永远不会被调用，因为它被提供了一个新的承诺，因为ctx.write(msg)本质上是ctx.channel.write(msg, newPromise())。

在我们的情况下，我们最终通过在所有业务逻辑处理程序之间注入相同的可共享错误处理来解决这种情况。

处理程序看起来像这样

@ChannelHandler.Sharable
class OutboundErrorHandler extends ChannelOutboundHandlerAdapter {

    private final static Logger logger = LoggerFactory.getLogger(OutboundErrorHandler.class);
    private Throwable handledCause = null;

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
        ctx.write(msg, promise).addListener(writeResult -> handleWriteResult(ctx, writeResult));
    }

    private void handleWriteResult(ChannelHandlerContext ctx, Future<?> writeResult) {
        if (!writeResult.isSuccess()) {
            final Throwable cause = writeResult.cause();

            if (cause instanceof ClosedChannelException) {
                // no reason to close an already closed channel - just ignore
                return;
            }

            // Since this handler is shared and added multiple times
            // we need to avoid spamming the logs N number of times for the same error
            if (handledCause == cause) return;
            handledCause = cause;

            logger.error("Uncaught exception on write!", cause);

            // By checking on channel writability and closing the channel after writing the error message,
            // only the first listener will signal the error to the client
            final Channel channel = ctx.channel();
            if (channel.isWritable()) {
                ctx.writeAndFlush(serverErrorJSON(cause.getMessage()), channel.newPromise());
                ctx.close();
            }
        }
    }
}

然后在我们的管道设置中，我们有这个

// Prepend the error handler to every entry in the pipeline. 
// The intention behind this is to have a catch-all
// outbound error handler and thereby avoiding the need to attach a
// listener to every ctx.write(...).
final OutboundErrorHandler outboundErrorHandler = new OutboundErrorHandler();
for (Map.Entry<String, ChannelHandler> entry : pipeline) {
    pipeline.addBefore(entry.getKey(), entry.getKey() + "#OutboundErrorHandler", outboundErrorHandler);
}

Answer 3

我找到了一个非常简单的解决方案，它允许入站和出站异常到达与管道中最后一个 ChannelHandler 相同的异常处理程序。

我的管道设置如下：

    //Inbound propagation
    socketChannel.pipeline()
      .addLast(new Decoder())
      .addLast(new ExceptionHandler());

    //Outbound propagation
    socketChannel.pipeline()
      .addFirst(new OutboundExceptionRouter())
      .addFirst(new Encoder());

这是我的 ExceptionHandler 的内容，它记录捕获的异常：

public class ExceptionHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("Exception caught on channel", cause);
    }
}

现在允许 ExceptionHandler 处理出站异常的魔法发生在 OutBoundExceptionRouter 中：

public class OutboundExceptionRouter extends ChannelOutboundHandlerAdapter {
    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        promise.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
        super.write(ctx, msg, promise);
    }
}

这是我的管道中调用的第一个出站处理程序，它的作用是向出站写入承诺添加一个侦听器，当承诺失败时，它将执行future.channel().pipeline().fireExceptionCaught(future.cause());。 fireExceptionCaught 方法通过管道向入站方向传播异常，最终到达 ExceptionHandler。

---

如果有人感兴趣，从 Netty 4.1 开始，我们需要添加一个监听器来获取异常的原因是因为在对通道执行 writeAndFlush 之后，invokeWrite0 method 在 AbstractChannelHandlerContext.java 中被调用，它包装了在 try catch 块中写入操作。 catch 块通知 Promise 而不是为入站消息调用 fireExceptionCaught like the invokeChannelRead method does。