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

在写这个文章前不得不吐槽目前国内一些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 的配置是匹配的。