如何在Spring Cloud Stream中为RabbitMQ消息动态设置路由键？

我需要将Kafka主题中的消息转发到多个RabbitMQ队列，计划用Kafka消息的键作为RabbitMQ消息的路由键，数据流转链路为：Kafka主题 -> Spring Cloud Stream应用 -> RabbitMQ。

RabbitMQ配置：

• Exchange类型为direct
• 所有队列均绑定至该Exchange，且路由键与队列名称完全一致

cloud:
  stream:
    default-binder: rabbit
    function:
      definition: upstreamProcessor
    bindings:
      upstreamProcessor-in-0:
        destination: 这里填写Kafka主题名称
        group: ms-upstream-provider
        binder: kafka
      upstreamProcessor-out-0:
        destination: 这里填写RabbitMQ Exchange名称
        binder: rabbit
    rabbit.bindings.upstreamProcessor-out-0.producer:
      exchangeType: direct

public Function<Message<String>, Message<JsonNode>> upstreamProcessor() {
    return message -> {
        try {
            JsonNode node = mapper.readTree(message.getPayload());
            String routingKey= message.getHeaders().get(KafkaHeaders.RECEIVED_KEY).toString();
            return MessageBuilder.withPayload(node)
                    .setHeader("routingKey", routingKey)
                    .build();
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    };
}

你当前设置的自定义routingKey header不会被RabbitMQ binder识别，需要使用Spring Cloud Stream RabbitMQ专属的Header来指定路由键，修改后的代码如下：

import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.kafka.support.KafkaHeaders;

private final ObjectMapper mapper;

public Function<Message<String>, Message<JsonNode>> upstreamProcessor() {
    return message -> {
        try {
            JsonNode node = mapper.readTree(message.getPayload());
            String routingKey = message.getHeaders().get(KafkaHeaders.RECEIVED_KEY).toString();
            return MessageBuilder.withPayload(node)
                    // 使用RabbitMQ binder识别的路由键Header
                    .setHeader(AmqpHeaders.ROUTING_KEY, routingKey)
                    .build();
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    };
}

关键说明 #

• AmqpHeaders.ROUTING_KEY是Spring AMQP提供的常量，对应header名称amqp_routingKey，RabbitMQ binder会自动读取该值作为消息的路由键
• 结合你direct类型Exchange、路由键与队列名称一致的配置，消息会自动路由到对应名称的队列
• 这种方式完全通过代码动态设置路由键，无需为每个队列配置绑定，适配你6000+队列的场景

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