Spring Boot Camel：如何用onException处理Kafka DisconnectException？

首先明确：你完全不需要Quarkus来使用onException()——这是Apache Camel核心框架的原生功能，Spring Boot环境下直接就能用，放心上手就行。

下面给你几个实用的实现方案，覆盖日志记录、自动重试、自定义告警等常见需求：

1. 基础方案：记录异常并标记为已处理 #

这个方案会捕获DisconnectException，记录详细日志，同时标记异常已处理，避免整个路由因单次连接中断而终止：

@Component
public class KafkaTopicService extends RouteBuilder { // 注意：是RouteBuilder，不是你代码里的RouteBilder

    @Override
    public void configure() throws Exception {
        // 定义异常处理规则
        onException(DisconnectException.class)
                .log(LoggingLevel.ERROR, "Kafka连接断开异常: ${exception.message}")
                .handled(true); // 标记异常已处理，路由会继续运行

        // 你的Kafka消费路由
        from("kafka:myTopic?brokers=localhost:9092")
                .log("Message received from Kafka: ${body}");
    }
}

2. 进阶方案：添加自动重试机制 #

Kafka断开往往是临时网络波动导致的，配置重试策略可以自动尝试恢复连接：

@Component
public class KafkaTopicService extends RouteBuilder {

    @Override
    public void configure() throws Exception {
        // 配置重发策略
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        redeliveryPolicy.setMaximumRedeliveries(3); // 最多重试3次
        redeliveryPolicy.setRedeliveryDelay(2000); // 首次重试间隔2秒
        redeliveryPolicy.setBackOffMultiplier(2); // 重试间隔逐次翻倍（2s→4s→8s）

        onException(DisconnectException.class)
                .log(LoggingLevel.WARN, "Kafka连接断开，正在重试（第${exchangeProperty.CamelRedeliveryCounter}次）: ${exception.message}")
                .redeliveryPolicy(redeliveryPolicy)
                .handled(true); // 重试完成后标记异常已处理

        from("kafka:myTopic?brokers=localhost:9092")
                .log("Message received from Kafka: ${body}");
    }
}

3. 自定义处理：发送告警或保存失败上下文 #

如果重试后仍无法恢复，你可以扩展自定义逻辑，比如发送运维告警、把失败的消息上下文存入数据库留待后续排查：

@Component
public class KafkaTopicService extends RouteBuilder {

    // 注入自定义的告警服务
    @Autowired
    private AlertService alertService;

    @Override
    public void configure() throws Exception {
        onException(DisconnectException.class)
                .log(LoggingLevel.ERROR, "Kafka连接最终失败，触发告警: ${exception.message}")
                .bean(alertService, "sendKafkaDisconnectAlert") // 调用告警服务
                .handled(true);

        from("kafka:myTopic?brokers=localhost:9092")
                .log("Message received from Kafka: ${body}");
    }
}

// 示例告警服务
@Service
public class AlertService {
    public void sendKafkaDisconnectAlert(Exception exception) {
        // 这里可以实现具体告警逻辑：比如调用邮件API、企业微信机器人、短信接口等
        System.err.println("[告警] Kafka连接异常：" + exception.getMessage());
    }
}

额外小提示 #

• 如果不想完全"吞掉"异常，而是希望路由继续运行的同时保留异常上下文，可以把.handled(true)改成.continued(true)
• 你还可以配合Camel Kafka组件的内置配置（比如session.timeout.ms、max.poll.interval.ms）一起优化连接稳定性

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