前言

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

本文介绍三种应用方式：

1：普通整合RabbitMQ

2：消息分区

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

1:核心依赖

<dependency>

    <groupId>org.springframework.cloud</groupId>

    <artifactId>spring-cloud-stream-binder-rabbit</artifactId>

    <version>${spring.cloud.stream}</version>

</dependency>

<dependency>

    <groupId>org.springframework.cloud</groupId>

    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>

    <version>${spring.cloud.stream}</version>

</dependency>

全部依赖：

项目结构图：

2:普通整合RabbitMQ

2.1:application.properties

spring.rabbitmq.host=192.168.1.218

spring.rabbitmq.port=5672

spring.rabbitmq.username=admin

spring.rabbitmq.password=123456

spring.cloud.stream.bindings.dev-exchange.destination=dev-exchange

spring.cloud.stream.bindings.dev-exchange.group=dev-queue

spring.cloud.stream.bindings.dev-exchange.content-type=application/json

spring.cloud.stream.bindings.dev-exchange.consumer.concurrency=1

spring.cloud.stream.bindings.dev-exchange.consumer.max-attempts=1

2.2:定义生产者和消费者接口

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();

}

2.3:生产者/消费者的具体实现

生产者代码：

@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);

    }

2.4:测试

发送请求：

控制台日志：

3:消息分区

3.1:概念

所谓消息分区就是将一个大队列拆分成多个小队列，然后分解成 producer-A -> queue-A -> Consumer-A 的一种场景。

3.2:如何在项目中使用

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

## RabbitMQ 消息分区配置

spring.cloud.stream.bindings.partition-exchange.destination=partition-exchange

spring.cloud.stream.bindings.partition-exchange.group=partition-queue

spring.cloud.stream.bindings.partition-exchange.content-type=application/json

spring.cloud.stream.bindings.partition-exchange.consumer.concurrency=1

spring.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-1

spring.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的队列。