SpringCloud Stream整合RabbitMQ3.5.0

科技 07-28 来源：奋斗喝咖啡

前言

本文章为单体项目，将消费者和生产者写在同一个项目中，介意者不用向下看了。

本文介绍三种应用方式：

1：普通整合RabbitMQ

2：消息分区

3：按条件消费（多个消费者只消费同一队列中满足自己条件的消息）

1：核心依赖

    org.springframework.cloud    spring-cloud-stream-binder-rabbit    ${spring.cloud.stream}    org.springframework.cloud    spring-cloud-starter-stream-rabbit    ${spring.cloud.stream}

全部依赖：

项目目录图：

2：基础版整合RabbitMQ

①:application.properties

spring.rabbitmq.host=192.168.1.218spring.rabbitmq.port=5672spring.rabbitmq.username=adminspring.rabbitmq.password=123456spring.cloud.stream.bindings.dev-exchange.destination=dev-exchangespring.cloud.stream.bindings.dev-exchange.group=dev-queuespring.cloud.stream.bindings.dev-exchange.content-type=application/jsonspring.cloud.stream.bindings.dev-exchange.consumer.concurrency=1spring.cloud.stream.bindings.dev-exchange.consumer.max-attempts=1

②：定义生产者和消费者接口

import com.boot.rabbitmq.constance.MQConstants;import org.springframework.cloud.stream.annotation.Input;import org.springframework.cloud.stream.annotation.Output;import org.springframework.messaging.MessageChannel;import org.springframework.messaging.SubscribableChannel;public interface RabbitStream {    /**     * 消息流入（消费）     **/    @Input(MQConstants.DEV_EXCHANGE)    SubscribableChannel devConsumer();    /**     * 消息流出（生产）     **/    @Output(MQConstants.DEV_EXCHANGE)    MessageChannel devProducer();}

③：生产者代码：

@Component@EnableBinding(RabbitStream.class)public class DevProducer {    private static final Logger logger = LoggerFactory.getLogger(DevProducer.class);    private final RabbitStream rabbitStream;    public DevProducer(RabbitStream rabbitStream) {        this.rabbitStream = rabbitStream;    }    public void sendMsg(MQModel model) {        logger.info("producer:{}", JSON.toJSONString(model));        rabbitStream.devProducer()                .send(MessageBuilder.withPayload(model).build());    }}

④：费者代码：

@Component@EnableBinding(RabbitStream.class)public class DevListener {    private static final Logger logger = LoggerFactory.getLogger(DevListener.class);    @StreamListener(MQConstants.DEV_EXCHANGE)    public void receiveMsgAutoCommit(@Payload String payload) {        logger.info("consumer:{}", payload);    }}

⑥：controller代码：

    @PostMapping(value = "/dev")    public void dev(@RequestBody MQModel model) {        devProducer.sendMsg(model);    }

⑦：测试

发送请求：

控制台日志：

3：消息分区

3.1：概念

RabbitMQ 本身是不支持消息分区的，只是由于业务演变+代码控制的一种方案而已（参考spring官方开头文档理解）。
个人理解：所谓消息分区就是将一个大队列拆分 0、1...n 个小队列，
然后分解成 producer-A -> queue-A -> Consumer-A 的一种场景。

3.2：如何在项目中使用

①：配置
不需要改很多东西，只需要添加少部分配置即可

## RabbitMQ 消息分区配置spring.cloud.stream.bindings.partition-exchange.destination=partition-exchangespring.cloud.stream.bindings.partition-exchange.group=partition-queuespring.cloud.stream.bindings.partition-exchange.content-type=application/jsonspring.cloud.stream.bindings.partition-exchange.consumer.concurrency=1spring.cloud.stream.bindings.partition-exchange.consumer.max-attempts=1## 消息分区spring.cloud.stream.bindings.partition-exchange.consumer.partitioned=true## 分区数量spring.cloud.stream.bindings.partition-exchange.producer.partition-count=2## 机器下标，最大值=partition-count-1spring.cloud.stream.instance-index=0## 分区策略表达式spring.cloud.stream.bindings.partition-exchange.producer.partition-key-expression=payload.mid

②：路由规则
然后消息的路由的时候会从payload拿到mid进行条件运算：
mid/2=1则放在应用队列下标为1的队列，mid/2=0则放在队列下标为0的队列。