本文详解如何在 Spring Integration 的 IntegrationFlow 中为 Kafka 出站通道适配器（Kafka.outboundChannelAdapter）配置 RetryAdvice，实现消息发布失败时的自动重试，避免因网络抖动或临时性 Broker 不可用导致的消息丢失。
本文详解如何在 spring integration 的 `integrationflow` 中为 kafka 出站通道适配器（`kafka.outboundchanneladapter`）配置 `retryadvice`，实现消息发布失败时的自动重试，避免因网络抖动或临时性 broker 不可用导致的消息丢失。
在基于 Spring Integration 构建的 Kafka 消息发布流程中，Kafka.outboundChannelAdapter 默认以异步方式发送消息（即 sync = false），此时异常不会向上传播至流中，导致 RetryAdvice 无法生效。要启用重试机制，必须显式启用同步发送模式（sync(true)），使异常可被捕获并交由 RetryAdvice 处理。
以下是完整、可运行的配置示例：
@Bean
public RetryAdvice retryAdvice() {
var retryTemplate = new RetryTemplate();
retryTemplate.setRetryPolicy(new SimpleRetryPolicy(3)); // 最多重试 3 次
retryTemplate.setBackOffPolicy(new FixedBackOffPolicy() {{
setBackOffPeriod(1000L); // 每次重试间隔 1 秒
}});
return new RequestHandlerRetryAdvice(retryTemplate);
}
@Bean
public IntegrationFlow publisherFlow(
CsvToEmailInteractionConverter converter,
KafkaTemplate<String, String> kafkaTemplate) {
return IntegrationFlow.from("publisherChannel")
.transform(converter, "transform")
.transform(Transformers.toJson())
.log(LoggingHandler.Level.DEBUG, Constants.LOG_FLOW_CATEGORY,
m -> "Payload: " + m.getPayload())
.handle(Kafka.outboundChannelAdapter(kafkaTemplate)
.topic(Constants.KAFKA_TOPIC)
.sync(true), // ⚠️ 关键：必须设为 true 才能触发重试
e -> e.id(Constants.ENGAGE_ACOUSTIC_ID)
.advice(retryAdvice())) // 绑定重试建议
.get();
}⚠️ 关键注意事项：
-
sync(true)是启用重试的前提——异步发送（默认）会吞掉异常，RetryAdvice将完全失效； -
RetryAdvice必须通过.advice(...)显式绑定到.handle()节点，不能作用于上游转换器（如transform）； - 若重试全部失败，异常将抛出至流的错误通道（如未配置，则传播为
MessageDeliveryException），建议配合.routeByException()进行兜底处理（例如记录告警、转入死信队列）； - 生产环境建议结合
StatefulRetryOperationsInterceptor或RecoveryCallback实现更精细的失败恢复逻辑（如降级写入数据库）。
综上，正确配置 sync(true) 与 RetryAdvice 的组合，即可为 Kafka 发布流程提供健壮的容错能力，显著提升系统可靠性。
