Cloud Stream无法追踪下游故障：RabbitMQ转Kafka消息防丢诉求

针对你用Spring Cloud Stream函数式实现RabbitMQ到Kafka消息流转时，Kafka宕机期间消息丢失、错误通道未触发、sync配置无效的问题，以下是具体解决方案：

一、修正错误通道监听配置（解决@ServiceActivator未执行） #

Cloud Stream的生产者错误通道命名规则为{输出绑定名}.errors，你的输出绑定是handle-out-0，因此错误通道应为handle-out-0.errors，而非自定义的error-topic。修改代码如下：

@ServiceActivator(inputChannel = "handle-out-0.errors")
public void errorHandler(ErrorMessage em) {
    log.info("捕获Kafka发送错误: {}", em);
    if (em.getPayload() instanceof KafkaSendFailureException) {
        KafkaSendFailureException kafkaEx = (KafkaSendFailureException) em.getPayload();
        log.warn("发送失败的消息内容: {}", kafkaEx.getRecord().value());
        // 这里可扩展将消息转发到RabbitMQ DLQ或其他持久化存储
    }
}

二、调整Kafka生产者配置，让sync模式生效 #

你当前设置的max.block.ms=100时间过短，导致生产者未触发阻塞就抛出异常，无法体现sync模式的阻塞等待效果。延长该值并确保sync配置正确：

spring:
  cloud:
    stream:
      bindings:
        handle-out-0:
          destination: mytopic
          producer:
            sync: true
            errorChannelEnabled: true
          binder: kafka
      kafka:
        binder:
          producer:
            properties:
              max.block.ms: 30000  # 延长至30秒，让sync模式真正生效

开启sync=true后，生产者会同步等待Kafka发送结果，若Kafka不可用则阻塞至超时，异常会被错误通道捕获。

三、实现Kafka宕机时的RabbitMQ消息处理策略 #

方案1：暂停RabbitMQ消费者（避免无效拉取） #

通过监听Kafka绑定的健康状态，自动暂停/恢复RabbitMQ消费者：

@Component
public class KafkaHealthListener {

    @Autowired
    private MessageChannel handleIn0;

    @EventListener
    public void onKafkaHealthChange(HealthStatusChangedEvent event) {
        if ("kafka".equals(event.getSource().getName())) {
            if (Status.DOWN.equals(event.getHealth().getStatus())) {
                ((SubscribableChannel) handleIn0).stop();
                log.info("Kafka不可用，已暂停RabbitMQ消费者");
            } else {
                ((SubscribableChannel) handleIn0).start();
                log.info("Kafka已恢复，重启RabbitMQ消费者");
            }
        }
    }
}

方案2：将失败消息转发到RabbitMQ DLQ #

开启RabbitMQ的死信队列转发，避免消息循环消费：

spring:
  cloud:
    stream:
      rabbit:
        bindings:
          handle-in-0:
            consumer:
              bindingRoutingKey: MyRoutingKey
              exchangeType: topic
              requeueRejected: false  # 关闭自动重入队
              acknowledgeMode: AUTO
              republishToDlq: true  # 开启自动转发到DLQ

同时在错误处理方法中手动拒绝消息，触发DLQ转发：

@ServiceActivator(inputChannel = "handle-out-0.errors")
public void errorHandler(ErrorMessage em) {
    Message<?> originalMsg = ((MessagingException) em.getPayload()).getFailedMessage();
    Channel channel = originalMsg.getHeaders().get(AmqpHeaders.CHANNEL, Channel.class);
    Long deliveryTag = originalMsg.getHeaders().get(AmqpHeaders.DELIVERY_TAG, Long.class);
    
    try {
        // 拒绝消息，触发DLQ转发（需配合republishToDlq=true）
        channel.basicReject(deliveryTag, false);
        log.info("消息已转发至RabbitMQ DLQ");
    } catch (IOException e) {
        log.error("拒绝消息失败", e);
    }
}

四、移除冲突的自定义Kafka配置 #

你自定义的KafkaProducerConfiguration会与Cloud Stream的Kafka Binder冲突，Binder会自动管理ProducerFactory和KafkaTemplate，直接删除该配置类即可。

五、基础保障：强化RabbitMQ消息确认 #

确保现有配置中的spring.rabbitmq.publisher-confirms: true和RabbitMQ消费者acknowledgeMode: AUTO生效，只有当消息成功发送到Kafka（或处理完成）后，才会确认RabbitMQ消息，避免丢失。

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