如何基于条件通过Spring Cloud Stream向2个Kafka Topic发布消息

方案1：使用动态发送目标头（最简实现） #

Spring Cloud Stream 内置支持通过消息头指定发送目标，直接覆盖配置文件中的默认输出topic，无需新增绑定配置：

• 修改apply方法的catch块逻辑，构造异常消息，添加spring.cloud.stream.sendto.destination头指定自定义异常topic即可，示例代码：

@Override
public Message<OutputMessage> apply(Message<InputMessage> inputMessage) {
  try {
    Message<OutputMessage> outputMessage = process(inputMessage);
    return outputMessage;
  } catch (Exception e) {
    OutputMessage errorMsg = buildErrorMsg(inputMessage, e); // 自行实现异常消息构造逻辑
    return MessageBuilder.withPayload(errorMsg)
        .setHeader("spring.cloud.stream.sendto.destination", "custom.error.topic") // 指定自定义异常topic
        .build();
  }
}

如果正常业务逻辑也需要按需发不同topic，同理在正常返回的消息头加该属性即可。

方案2：使用StreamBridge主动发送（最灵活，支持多topic同时发送） #

如果需要同一次请求向多个topic发布消息，或者需要完全控制发送逻辑，直接注入StreamBridge手动发送消息即可：

1. 改造KafkaTransformer类，注入StreamBridge：

public class KafkaTransformer implements Function<Message<InputMessage>, Message<OutputMessage>> {

  @Autowired
  private StreamBridge streamBridge;

  @Override
  public Message<OutputMessage> apply(Message<InputMessage> inputMessage) {
    try {
      Message<OutputMessage> outputMessage = process(inputMessage);
      // 示例：正常流程额外发送到统计topic
      streamBridge.send("statistics.topic", outputMessage);
      return outputMessage;
    } catch (Exception e) {
      // 异常场景直接发送到自定义异常topic
      OutputMessage errorMsg = buildErrorMsg(inputMessage, e);
      streamBridge.send("custom.error.topic", errorMsg);
      // 不需要默认输出返回null即可，也可以选择返回消息走默认绑定的output.topic
      return null;
    }
  }
}

2. 不需要额外修改配置文件，原有配置保持不变即可。

方案3：固定多输出绑定（适合输出topic数量固定的场景） #

如果输出topic数量固定，也可以通过返回多元素Tuple实现，需要配套修改绑定配置：

1. 修改函数返回类型为Tuple2：

@Bean
public Function<Message<InputMessage>, Tuple2<Message<OutputMessage>, Message<OutputMessage>>> messageTransformer(){
    return new KafkaTransformer();
}

2. 配置文件新增第二个输出绑定：

spring.cloud.stream.bindings.messageTransformer-out-0.destination=output.topic
spring.cloud.stream.bindings.messageTransformer-out-1.destination=custom.error.topic

3. 方法返回时，对应位置放对应消息，无消息返回null即可。

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