SpringCloud Stream生产者配置RabbitMq的动态路由键

时间:2022-01-09 23:38:28

在写这个文章前不得不吐槽目前国内一些blog的文章,尽是些复制粘贴的文章,提到点上但没任何的深入和例子。.........

经过测试下来总结一下RabbitMQ的Exchange的特性:

1、direct

生产者可以指定路由键,消费者可以指定路由键,但不能讲路由键设置为#(全部)。

2、topic

生产者可以指定路由键,消费者可以指定路由键,也可以不指定(或者是#)。

3、fanout

生产者和消费都忽略路由键。

在现实的场景里,通常是生产者会生产多个路由键的消费,然后多个消费来消费指定路由键的消息,但通常生产者的生产代码是同一份,如何在发消息的时候动态指定当前消息的路由键呢?

例子:门店平台系统集中处理多个门店的数据,然后分别将不同门店的数据发送到不同的门店(即:A门店只消费属于A门店的消息,B门店只消费属于B的消息)

看例子:

application.yml

 spring:
cloud:
stream:
# 设置默认的binder
default-binder: pos
binders:
scm:
type: rabbit
environment:
spring:
rabbitmq:
# 连接到scm的host和exchange
virtual-host: scm
pos:
type: rabbit
environment:
spring:
rabbitmq:
# 连接到pos的host和exchange
virtual-host: pos shop:
type: rabbit
environment:
spring:
rabbitmq:
# 连接到shop的host和exchange
virtual-host: shop bindings:
# ---------消息消费------------ # 集单开始生产消费
order_set_start_produce_input:
binder: pos
destination: POS_ORDER_SET_STRAT_PRODUCE
group: pos_group # 门店ID为1的消费者
shop_consumer_input_1:
binder: shop
destination: POS_ORDER_SET_STRAT_PRODUCE
group: shop_1_group #-----------消息生产-----------
# 集单开始生产通知生产
order_set_start_produce_output:
binder: pos
destination: POS_ORDER_SET_STRAT_PRODUCE rabbit:
bindings:
# 集单开始生产消费者
order_set_start_produce_input:
consumer:
exchangeType: topic
autoBindDlq: true
republishToDlq: true
deadLetterExchange: POS_ORDER_SET_STRAT_PRODUCE_POS_DLX
#bindingRoutingKey: '#'
# 门店1的消费者
shop_consumer_input_1:
consumer:
exchangeType: topic
autoBindDlq: true
republishToDlq: true
deadLetterExchange: POS_ORDER_SET_STRAT_PRODUCE_SHOP_1_DLX
bindingRoutingKey: 1
70 deadLetterRoutingKey: 1 # 生产者配置
order_set_start_produce_output:
producer:
exchangeType: topic
76 routingKeyExpression: headers.shopId
# routingKeyExpression: headers['shopId']

上面的配置文件配置了一个动态的基于shopId做路由的生产者配置,一个消费全部路由键的消费者,如果要配置指定路由键的可以在配置文件里设置bindingRoutingKey属性的值。

生产者java代码:

import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder; import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.longge.pos.production.mq.dto.OrderSetProductionMsg; import lombok.extern.slf4j.Slf4j; @Slf4j
public class MqSendUtil {
private static MessageChannel orderSetStartProduceChannel; public static void setSfOrderCreateChannel(MessageChannel channel) {
sfOrderCreateProduceChannel = channel;
} public static void sendOrderSetPrintMsg(OrderSetProductionMsg msg) {
// add kv pair - routingkeyexpression (which matches 'type') will then evaluate
// and add the value as routing key
log.info("发送开始生产的MQ:{}", JSONObject.toJSONString(msg));
orderSetStartProduceChannel.send(MessageBuilder.withPayload(JSON.toJSONString(msg)).setHeader("shopId", msg.getOrderSet().getShopId()).build());
//orderSetStartProduceChannel.send(MessageBuilder.withPayload(JSON.toJSONString(msg)).build());
}
}

动态路由的核心在于上面那个红色的字体的地方,这个是和配置文件里的  routingKeyExpression 的配置是匹配的。

SpringCloud Stream生产者配置RabbitMq的动态路由键