Spring Cloud Stream生产者PublishConfirm配置方法咨询（Supplier/StreamBridge）

Spring Cloud Stream本身是消息中间件的抽象层，PublishConfirm的具体配置是通过绑定器传递给底层中间件（比如RabbitMQ、Kafka）的，但它提供了统一的配置入口和事件机制来处理确认逻辑，下面分别说明Supplier和StreamBridge两种方式的配置方法：

一、Supplier方式的配置 #

• Supplier的PublishConfirm配置主要靠绑定属性实现，这些属性会被传递给对应的底层绑定器。
• 举个RabbitMQ的配置例子，在application.yml里这么写：

spring:
  cloud:
    stream:
      bindings:
        supplier-out-0: # Supplier默认的输出绑定名称
          destination: your-target-topic
          producer:
            confirm-type: correlated # 开启关联式确认
      rabbit:
        bindings:
          supplier-out-0:
            producer:
              mandatory: true # 确保消息无法路由时触发回调

如果是Kafka，就换成Kafka对应的参数：

spring:
  cloud:
    stream:
      bindings:
        supplier-out-0:
          destination: your-target-topic
          producer:
            acks: all # 要求所有副本确认后才算发送成功

• 要处理确认结果的话，不用针对中间件写特定代码，直接监听SCSt提供的ProducerAcknowledgementEvent事件就行：

@EventListener
public void handleProducerAck(ProducerAcknowledgementEvent event) {
    Message<?> originalMsg = event.getOriginalMessage();
    boolean isSuccess = event.isAcknowledged();
    // 这里写你的确认逻辑，比如记录日志、更新消息状态
}

二、StreamBridge方式的配置 #

StreamBridge是动态发送消息的方式，配置PublishConfirm有两种方式：

1. 全局默认配置：直接在全局生产者属性里设置，所有StreamBridge发送的消息都会默认用这个配置：

spring:
  cloud:
    stream:
      producer:
        confirm-type: correlated # RabbitMQ示例，Kafka用acks: all

2. 动态指定配置：发送单条消息时，可以通过消息头或者发送请求参数单独设置：

// 方式1：通过消息头传递配置
Map<String, Object> headers = new HashMap<>();
headers.put(RabbitProducerHeaders.CONFIRM_TYPE, "correlated"); // RabbitMQ专属
headers.put(KafkaHeaders.ACKS, "all"); // Kafka专属
Message<String> msg = MessageBuilder.createMessage("test content", new MessageHeaders(headers));
streamBridge.send("your-destination", msg);

// 方式2：用SendRequest指定生产者属性
StreamBridge.SendRequest sendRequest = StreamBridge.SendRequest.builder()
    .destination("your-destination")
    .message(msg)
    .producerProperties(Map.of("confirm-type", "correlated"))
    .build();
streamBridge.send(sendRequest);

• 和Supplier一样，确认结果也是通过监听ProducerAcknowledgementEvent来处理，事件里包含原始消息和确认状态。

重要提示 #

Spring Cloud Stream本身不定义PublishConfirm的具体规则，只是把配置转交给底层绑定器。文档里多拿AMQP举例是因为RabbitMQ的确认机制比较典型，但这套配置思路对其他绑定器（比如Kafka）同样适用，只是具体参数名不一样而已。不管用Supplier还是StreamBridge，核心都是通过绑定属性配置确认规则，通过统一的事件监听处理确认结果。

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