Spring Boot中如何捕获并记录Kafka监听器与@Scheduled方法的异常？

针对Kafka监听器、异步@EventListener、@Scheduled方法中未显式捕获的异常，可通过以下几种方式统一捕获并记录日志：

1. Kafka监听器异常处理 #

实现ConsumerAwareListenerErrorHandler接口自定义异常处理器，在处理器中完成日志记录和异常逻辑处理，然后通过@KafkaListener的errorHandler属性绑定该处理器：

@Component
public class CustomKafkaErrorHandler implements ConsumerAwareListenerErrorHandler {
    private static final Logger logger = LoggerFactory.getLogger(CustomKafkaErrorHandler.class);

    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {
        logger.error("Kafka监听器处理消息时发生异常，消息内容: {}", message.getPayload(), exception);
        // 可根据业务需求添加重试、死信队列投递等逻辑
        return null;
    }
}

使用时指定处理器：

@KafkaListener(topics = "your-topic", errorHandler = "customKafkaErrorHandler")
public void listen(String message) {
    // 业务逻辑，无需显式捕获异常
}

2. 异步任务（含异步@EventListener）全局异常处理 #

实现AsyncConfigurer接口，自定义异步任务的未捕获异常处理器，注册到@EnableAsync的配置类中：

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {
    private static final Logger logger = LoggerFactory.getLogger(AsyncConfig.class);

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (throwable, method, params) -> {
            logger.error("异步方法执行异常，方法名: {}, 参数: {}", method.getName(), Arrays.toString(params), throwable);
        };
    }
}

注：给@EventListener添加@Async注解后，该处理器会自动捕获其抛出的未处理异常。

3. @Scheduled任务全局异常处理 #

通过SchedulingConfigurer自定义任务调度器，设置ErrorHandler来捕获调度任务中的未处理异常：

@Configuration
@EnableScheduling
public class SchedulerConfig implements SchedulingConfigurer {
    private static final Logger logger = LoggerFactory.getLogger(SchedulerConfig.class);

    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        taskRegistrar.setScheduler(customTaskScheduler());
    }

    @Bean
    public TaskScheduler customTaskScheduler() {
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setErrorHandler(throwable -> logger.error("调度任务执行异常", throwable));
        scheduler.setThreadNamePrefix("scheduled-task-pool-");
        scheduler.initialize();
        return scheduler;
    }
}

4. 全局线程未捕获异常兜底处理 #

如果以上针对性配置仍有遗漏，可通过自定义线程池并设置UncaughtExceptionHandler作为兜底方案，覆盖所有线程执行时的未捕获异常：

@Bean
public Executor globalTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setThreadNamePrefix("global-task-pool-");
    executor.setUncaughtExceptionHandler((thread, throwable) -> 
        logger.error("线程[{}]执行时发生未捕获异常", thread.getName(), throwable)
    );
    executor.initialize();
    return executor;
}

内容的提问来源于stack exchange，提问作者John Little