快捷導(dǎo)航

SpringCloud Stream RabbitMQ動(dòng)態(tài)路由Key問(wèn)題

更新時(shí)間：2026年05月06日 09:17:26 作者：這個(gè)饕字怎么讀

文章主要描述了使用消息中間件實(shí)現(xiàn)不同操作后發(fā)送不同郵件的功能,通過(guò)設(shè)置路由key和group,使得生產(chǎn)者能夠根據(jù)郵件類型發(fā)送消息至對(duì)應(yīng)隊(duì)列,消費(fèi)者則通過(guò)路由key篩選消息

前言

這里有個(gè)業(yè)務(wù)是這樣的，我需要在不同的操作后給用戶發(fā)送不同的郵件，由于比較耗時(shí)所以引入消息中間件，不同的郵件對(duì)應(yīng)的消息類型是不一樣的，所以需要生產(chǎn)者往隊(duì)列里發(fā)送數(shù)據(jù)時(shí)綁定好路由key，例如：

圖里表示交換機(jī)根據(jù)路由key綁定了不同的隊(duì)列。

要達(dá)到這種效果，首先消費(fèi)者肯定是可以根據(jù)路由key來(lái)決定消息是不是發(fā)送給自己的，對(duì)于生產(chǎn)者則需要用到routingKeyExpression 來(lái)決定往哪個(gè)路由key發(fā)送數(shù)據(jù)（大概是這個(gè)意思）。

然后就是stream中的group其實(shí)對(duì)應(yīng)到rabbitMQ中就是隊(duì)列的概念，所以我們這里設(shè)置兩個(gè)不同的group來(lái)對(duì)應(yīng)到不同的隊(duì)列，區(qū)分開(kāi)業(yè)務(wù)；

例子

這里我定義了兩個(gè)服務(wù)對(duì)應(yīng)消費(fèi)者和生產(chǎn)者。

生產(chǎn)者

spring:
  application:
    name: producer
  cloud:
    stream:
      binders: # 綁定MQ服務(wù)信息（此處我們是RabbitMQ）
        etpmsRabbitMQ: # 給Binder定義的名稱，?于后?的關(guān)聯(lián)
          type: rabbit # MQ類型，如果是Kafka的話，此處配置kafka
          environment: # MQ環(huán)境配置（?戶名、密碼等）
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: admin
                password: xxxxxx
      bindings: # 關(guān)聯(lián)整合通道和binder對(duì)象
        output: # output是我們定義的通道名稱，此處不能亂改
          destination: testExchange # 要使?的Exchange名稱（消息隊(duì)列主題名稱）
          content-type: text/plain # application/json # 消息類型設(shè)置，?如json
          binder: etpmsRabbitMQ # 關(guān)聯(lián)MQ服務(wù)
      rabbit:
        bindings:
          output:
            producer:
              # 生產(chǎn)者配置RabbitMq的動(dòng)態(tài)路由鍵
              routingKeyExpression: headers.type

package top.chenyt.producer;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

/**
 * @author yantao.chen
 */
@Service
public class ProviderService {
    /**
     * 將MessageChannel的封裝對(duì)象Source注?到這?使?
     */
    @Autowired
    private Source source;

    public void sendMessage(String content, String type) {
        // 向mq中發(fā)送消息（并不是直接操作mq，應(yīng)該操作的是spring cloud stream）
        // 使?通道向外發(fā)出消息(指的是Source??的output通道)
        source.output().send(MessageBuilder.withPayload(content).setHeader("type",type).build());
    }
}

package top.chenyt;

import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;

/**
 * @ClassName etpms-parent
 * @Author Jinondo
 * @Date 2022/1/31 12:42
 */
@SpringBootApplication
@Slf4j
@EnableBinding({Source.class})
public class ProducerApplication {

    public static void main(String[] args)  {
        SpringApplication.run(ProducerApplication.class, args);
    }

}

主要是yml配置添加：routingKeyExpression: headers.type

發(fā)送消息的時(shí)候setHeader一下

消費(fèi)者

spring:
  application:
    name: consumer
  cloud:
    stream:
      binders: # 綁定MQ服務(wù)信息（此處我們是RabbitMQ）
        etpmsRabbitMQ: # 給Binder定義的名稱，?于后?的關(guān)聯(lián)
          type: rabbit # MQ類型，如果是Kafka的話，此處配置kafka
          environment: # MQ環(huán)境配置（?戶名、密碼等）
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: admin
                password: xxxxx
      bindings: # 關(guān)聯(lián)整合通道和binder對(duì)象
        input: # input是我們定義的通道名稱，此處不能亂改
          destination: testExchange # 要使?的Exchange名稱（消息隊(duì)列主題名稱）
          content-type: text/plain # application/json # 消息類型設(shè)置，?如json,自動(dòng)將對(duì)象轉(zhuǎn)為json
          binder: etpmsRabbitMQ # 關(guān)聯(lián)MQ服務(wù)
          group: register
        my-input:
          destination: testExchange # 要使?的Exchange名稱（消息隊(duì)列主題名稱）
          content-type: text/plain # application/json # 消息類型設(shè)置，?如json,自動(dòng)將對(duì)象轉(zhuǎn)為json
          binder: etpmsRabbitMQ # 關(guān)聯(lián)MQ服務(wù)
          group: task
      rabbit:
        bindings:
          my-input:
            consumer:
              bindingRoutingKey: task
          input:
            consumer:
              bindingRoutingKey: register

這里我就定義了兩個(gè)通道，一個(gè)是默認(rèn)的input，一個(gè)是自定的

package top.chenyt.consumer;

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;

public interface MySink {

    String MY_INPUT = "my-input";

    @Input(MY_INPUT)
    SubscribableChannel myinput();

}

package top.chenyt.consumer;

import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;

/**
 * @ClassName etpms-parent
 * @Author Jinondo
 * @Date 2022/1/31 12:42
 */
@Service
public class ConsumerMsg {

    @StreamListener(Sink.INPUT)
    public void receiveMessages(Message<String> message) {
        System.out.println("========= input接收到的消息：" + message.getPayload());
    }

    @StreamListener(MySink.MY_INPUT)
    public void receiveMessages02(Message<String> message) {
        System.out.println("========= myinput接收到的消息：" + message.getPayload());
    }
}

package top.chenyt;

import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Sink;
import top.chenyt.consumer.MySink;

/**
 * @ClassName etpms-parent
 * @Author Jinondo
 * @Date 2022/1/31 12:42
 */
@SpringBootApplication
@Slf4j
@EnableBinding({Sink.class, MySink.class})
public class ConsumerApplication {

    public static void main(String[] args)  {
        SpringApplication.run(ConsumerApplication.class, args);
    }

}

這樣就能實(shí)現(xiàn)根據(jù)不同的消息類型對(duì)應(yīng)到不同的隊(duì)列且不同的路由key去了