Reactor 错误处理

• 引言
• 实现 OperatorErrorFunction
  • 代码示例
• 注册错误处理钩子
  • 代码示例
• 示例日志输出解析
• 总结
• 获取 event 错误信息

在使用 Reactor 框架开发响应式应用时，错误处理是一个关键的环节。Reactor 提供了多种机制来捕获和处理操作符抛出的异常，包括全局钩子（Hooks）和局部钩子（Context）。本文将通过示例代码详细介绍如何实现和注册错误处理钩子，并解释实际运行时的日志输出。

引言

在响应式编程中，错误处理不仅仅是捕获异常，还涉及到如何优雅地处理这些异常以保持应用的稳定性和可维护性。Reactor 提供了 Hooks 类，允许开发者在全局或局部范围内注册错误处理逻辑，从而实现对操作符抛出异常的统一处理。

实现 OperatorErrorFunction

OperatorErrorFunction 是一个实现了 BiFunction 接口的类，用于处理操作符抛出的异常。它接收一个异常和一个相关的数据对象，并返回一个新的异常。通过这种方式，开发者可以在异常传播到订阅者之前，对其进行日志记录、包装或其他处理。

代码示例

package nexus.io.telegram.bots.config;

import java.util.function.BiFunction;

import lombok.extern.slf4j.Slf4j;

/**
 * 这是一个实现 BiFunction 接口的类，用于处理操作符抛出的异常。
 * 它接收一个异常和一个相关的数据对象，并返回一个新的异常。
 */
@Slf4j
public class OperatorErrorFunction implements BiFunction<Throwable, Object, Throwable> {

  /**
   * 处理异常的方法。
   *
   * @param throwable 原始异常
   * @param data      相关的数据对象
   * @return 处理后的异常
   */
  @Override
  public Throwable apply(Throwable throwable, Object data) {
    log.error("error", throwable);
    return throwable;
  }
}

说明：

• @Slf4j 注解：使用 Lombok 提供的 @Slf4j 注解，自动为类生成一个 log 对象，用于记录日志。
• apply 方法：实现 BiFunction 接口的 apply 方法，该方法接收两个参数：
  • Throwable throwable：操作符抛出的原始异常。
  • Object data：与异常相关的数据对象（例如，导致异常的值）。
• 日志记录：在 apply 方法中，使用 log.error 记录异常信息。
• 异常返回：当前示例直接返回原始异常，可以根据需求对异常进行包装或转换。

注册错误处理钩子

在主应用程序中，需要将 OperatorErrorFunction 注册到 Reactor 的全局钩子中，以便在操作符抛出异常时调用该函数进行处理。

代码示例

package nexus.io.telegram.bots;

import nexus.io.annotation.AComponentScan;
import nexus.io.hotswap.wrapper.tio.boot.TioApplicationWrapper;
import nexus.io.telegram.bots.config.OperatorErrorFunction;

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Hooks;

@AComponentScan
@Slf4j
public class TelegramBotApp {
  public static void main(String[] args) {
    long start = System.currentTimeMillis();

    // 创建 OperatorErrorFunction 实例
    OperatorErrorFunction operatorErrorFunction = new OperatorErrorFunction();

    // 注册全局 onOperatorError 钩子
    Hooks.onOperatorError(operatorErrorFunction);

    // 取消注册全局钩子（在应用关闭时）
    //Hooks.resetOnOperatorError();

    // 注册全局 onErrorDropped 钩子
    Hooks.onErrorDropped((e) -> {
      log.error("hooks onErrorDropped:", e);
    });

    // 启动应用程序
    TioApplicationWrapper.run(TelegramBotApp.class, args);

    long end = System.currentTimeMillis();
    System.out.println((end - start) + "ms");
  }
}

说明：

• Hooks.onOperatorError：注册全局的 onOperatorError 钩子，传入 OperatorErrorFunction 实例。当操作符抛出异常时，该钩子将被调用。
• Hooks.onErrorDropped：注册全局的 onErrorDropped 钩子，用于处理那些未被订阅者捕获的异常。
• TioApplicationWrapper.run：启动应用程序，这里假设使用了 Tio 框架进行应用包装。

注意：

• 在应用关闭时，可以调用 Hooks.resetOnOperatorError() 来取消注册全局钩子，以避免潜在的内存泄漏。
• 注释掉的 Hooks.resetOnOperatorError() 方法在应用关闭时使用。

示例日志输出解析

以下是运行应用时的日志输出示例：

2024-12-11 00:15:41.108 [t4j-events-2] INFO   c.l.t.d.TelegramBotEventDispather.adapte:32 - message content:/start
2024-12-11 00:15:41.112 [t4j-events-2] ERROR  c.l.t.b.c.OperatorErrorFunction.apply:23 - error
java.lang.RuntimeException: File not found : "/hutujiqiren_welcome.html"
	at com.jfinal.template.source.FileSource.getContent(FileSource.java:68)
	at com.jfinal.template.Engine.buildTemplateBySourceFactory(Engine.java:210)
	at com.jfinal.template.Engine.getTemplate(Engine.java:195)
	at nexus.io.telegram.bots.command.HutuJiqiRenStartCommandFunction.apply(HutuJiqiRenStartCommandFunction.java:49)
	at nexus.io.telegram.command.TelegramCommandDispatcher.dispatch(TelegramCommandDispatcher.java:17)
	at nexus.io.telegram.dispather.TelegramBotEventDispather.lambda$0(TelegramBotEventDispather.java:39)
	at java.base/java.util.Optional.map(Optional.java:260)
	at nexus.io.telegram.dispather.TelegramBotEventDispather.adapte(TelegramBotEventDispather.java:39)
	at nexus.io.telegram.bots.adapter.MyBotEventAdapter.onSendMessage(MyBotEventAdapter.java:58)
	at telegram4j.core.event.EventAdapter.hookOnEvent(EventAdapter.java:115)
	at telegram4j.core.event.EventDispatcher.lambda$on$0(EventDispatcher.java:55)
	at reactor.core.publisher.FluxDefer.subscribe(FluxDefer.java:46)
	at reactor.core.publisher.Flux.subscribe(Flux.java:8660)
	at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.subscribeNext(MonoIgnoreThen.java:263)
	at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:51)
	at reactor.core.publisher.Mono.subscribe(Mono.java:4444)
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.onNext(FluxFlatMap.java:426)
	at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.runAsync(FluxPublishOn.java:440)
	at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.run(FluxPublishOn.java:527)
	at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84)
	at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:833)
2024-12-11 00:15:41.113 [t4j-events-2] ERROR  t.c.e.EventDispatcher.error:319 - Error while handling SendMessageEvent on nexus.io.telegram.bots.adapter.MyBotEventAdapter

日志解析：

1. INFO 日志：

   2024-12-11 00:15:41.108 [t4j-events-2] INFO   c.l.t.d.TelegramBotEventDispather.adapte:32 - message content:/start

   • 描述：接收到用户发送的 /start 消息。
   • 位置：TelegramBotEventDispather.adapte 方法第 32 行。

2. ERROR 日志：

   2024-12-11 00:15:41.112 [t4j-events-2] ERROR  c.l.t.b.c.OperatorErrorFunction.apply:23 - error
   java.lang.RuntimeException: File not found : "/hutujiqiren_welcome.html"
   ...

   • 描述：OperatorErrorFunction 在处理异常时记录了一条错误日志，并返回了原始异常。
   • 异常类型：RuntimeException: File not found : "/hutujiqiren_welcome.html"
   • 堆栈跟踪：显示了异常发生的具体位置，表明在模板引擎尝试加载 /hutujiqiren_welcome.html 文件时未找到该文件。
   • 位置：OperatorErrorFunction.apply 方法第 23 行。

3. ERROR 日志：

   2024-12-11 00:15:41.113 [t4j-events-2] ERROR  t.c.e.EventDispatcher.error:319 - Error while handling SendMessageEvent on nexus.io.telegram.bots.adapter.MyBotEventAdapter

   • 描述：在处理 SendMessageEvent 时发生错误。
   • 位置：EventDispatcher.error 方法第 319 行。

总结：

• 当 HutuJiqiRenStartCommandFunction 类中的 apply 方法尝试加载模板文件 /hutujiqiren_welcome.html 时，文件未找到，抛出了 RuntimeException。
• 该异常被 OperatorErrorFunction 捕获并通过 Hooks.onOperatorError 全局钩子进行处理，记录了错误日志。
• 此外，Hooks.onErrorDropped 钩子也被注册，用于处理那些未被订阅者捕获的异常。

总结

通过本文的介绍，已经了解了如何在 Reactor 中实现和注册错误处理钩子，以捕获和处理操作符抛出的异常。关键步骤包括：

1. 实现 BiFunction 接口的 OperatorErrorFunction 类：用于自定义异常处理逻辑。
2. 在主应用程序中注册全局钩子：使用 Hooks.onOperatorError 和 Hooks.onErrorDropped 方法，将自定义的错误处理逻辑应用到整个应用程序中。
3. 解析实际运行时的日志输出：理解错误处理钩子在实际运行中的作用和效果。

通过合理使用 Reactor 提供的错误处理机制，可以构建更加健壮和可维护的响应式应用，确保在各种异常情况下都能进行适当的处理，提升应用的稳定性和用户体验。

获取 event 错误信息

telegram4j.core.event.EventDispatcher

log.error("Error while handling {} on {}",event.getClass().getSimpleName(), adapter.getClass().getCanonicalName());