Spring Cloud Stream如何为每条消息指定Topic以路由至对应消费者？

你提到的核心困惑其实是对Spring Cloud Stream中消息路由的概念理解，以及如何自定义单条消息的路由目标。结合你提到的Exchange（应该是RabbitMQ场景），我来具体拆解：

先澄清概念：你说的"Topic"对应什么？ #

在Spring Cloud Stream + RabbitMQ的组合里，你口中的"Topic"其实对应消息的routing key——RabbitMQ的Exchange正是通过这个routing key来把消息路由到匹配的消费者队列的。而Spring Cloud Stream里的destination配置项，对应的是RabbitMQ中的Exchange名称（默认是topic类型的Exchange）。

之前你用source.output().send(MessageBuilder.withPayload(myMessage).build())无法指定路由规则，是因为默认情况下，Spring Cloud Stream会用绑定的destination作为默认的routing key，不会自定义单条消息的路由目标。

具体实现：为单条消息设置routing key #

要给每条消息指定自定义的路由规则，只需要在构建消息时，通过MessageBuilder设置amqp_routingKey这个header即可。示例代码如下：

import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

// 构建自定义路由的消息
String myMessage = "用户创建成功的通知内容";
Message<String> message = MessageBuilder
    .withPayload(myMessage)
    // 这里设置的就是供Exchange路由的routing key（即你说的Topic）
    .setHeader("amqp_routingKey", "user.created")
    .build();

// 发送消息
source.output().send(message);

消费者端的匹配配置 #

要让消费者能收到这条消息，你需要在消费者的绑定配置中，指定匹配的routing key规则。比如在application.yml中配置：

spring:
  cloud:
    stream:
      bindings:
        # 消费者的输入绑定
        input:
          # 必须和生产者的destination一致，对应同一个Exchange
          destination: user-exchange
          group: user-service-group
          consumer:
            # 匹配所有以user开头的routing key，比如user.created、user.updated
            routing-key-pattern: "user.*"
        # 生产者的输出绑定
        output:
          destination: user-exchange

如果你用的是Kafka场景（补充） #

如果是Kafka环境，你要给单条消息指定Topic的话，需要设置kafka_topic这个header，示例：

Message<String> message = MessageBuilder
    .withPayload(myMessage)
    .setHeader("kafka_topic", "user-created-topic")
    .build();
source.output().send(message);

纠正你的理解偏差 #

Spring Cloud Stream的设计是通过destination来绑定全局的Exchange（RabbitMQ）或Topic（Kafka），而单条消息的路由目标（RabbitMQ的routing key、Kafka的动态Topic）是通过消息header来动态设置的，不是直接修改全局的Topic配置。

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