【SpringBoot】SpringBoot 集成 RabbitMQ（一）（简单模式）

矫情吗；* 2023-10-02 09:42 5阅读 0赞

接下来使用 SpringBoot 来集成 RabbitMQ

### 1. SpringBoot 来集成 RabbitMQ 入门 ###

环境配置：

> JDK 1.8  
> IDEA 2019.1  
> SpringBoot 2.5.1  
> RabbitMQ 3.7.8  
> Erlang 21.1

工程结构图：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAc2NvNTI4Mg_size_11_color_FFFFFF_t_70_g_se_x_16_pic_left]

第一步》、新建一个 SpringBoot 工程，添加 POM 依赖：

<dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
        <scope>test</scope>
    </dependency>
    
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>

第二步》、添加 application.yml 文件配置

server:
      port: 8080
    spring:
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest
        password: guest
        virtual-host: /

上面两步执行完后，就得创建生产者和消费者了。

> 那么是先创建生产者，还是消费者呢？

如果先创建好生产者，然后向队列发送消息，如果该队列提前不存在的话，消息会发送失败；如果先创建消费者，让消费者提前创建好队列后，监听队列，再创建生产者，向队列中发送消息。

第三步》、创建消费者

**在消费端声明队列**

@Component
    @RabbitListener(queuesToDeclare = @Queue("simple_queue"))  //如果simple_queue队列不存在，则创建simple_queue队列。默认队列是持久化，非独占式的
    public class SimpleConsumer {
        
    
        // //消费者如果监听到消息队列有消息传入，则会自动消费
        @RabbitHandler
        public void receive(String message) {
        
            System.out.println("消费者接收到的消息是：" + message);
        }
    }

@RabbitListener 注解的属性的作用：

*  **queuesToDeclare**：如果 simple\_queue 队列不存在，则会自动创建simple\_queue队列。默认队列是持久化，非独占式的
 *  **queues**：里面的队列必须存在，否则就会报错。如：`@RabbitListener(queues = {"simple_queue2"})`，如果队列 simple\_queue2 不存在，那么启动消费者就会报错

第四步》、创建生产者

这里使用单元测试的方式创建，简单、方便

@SpringBootTest
    @RunWith(SpringRunner.class)
    public class AppTest {
        
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Test
        public void sendMsg() {
        
            String queueName = "simple_queue";
            String message = "Hello RabbitMQ SimpleQueue";
            rabbitTemplate.convertAndSend(queueName, message);
        }
    }

第五步》、启动单元测试方法

执行 `sendMsg()` 方法，它会启动整个 SpringBoot 项目，将消费者 SimpleConsumer 注入 Spring 容器中，监听队列 `simple_queue`（如果没有此队列，则会创建它）。然后，生产者就会发送消息，被消费者消费。

### 2. 新建一个生产者工程 ###

入门案例是将生产者放入了一个单元测试方法中，一启动单元测试方法，就会发送消息，然后立马被消费者消息了。这样，在 RabbitMQ 中看不到消息。

接下来将生产者放入一个单独的工程中，将它和消费者进行分离。

生产者工程图：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAc2NvNTI4Mg_size_13_color_FFFFFF_t_70_g_se_x_16]

第一步》、新建一个 SpringBoot 工程，添加 POM 依赖

同上

第二步》、添加 application.yml 配置

同上

【注意】：如果两个工程同时启动，则需要修改 `server.port` 号，这个值不能重复

第三步》、创建生产者

@Component
    public class SimpleProducer {
        
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void sendMsg() {
        
            for (int i = 0; i < 5; i++) {
        
                String msg = "简单模式下的消息 " + i;
                rabbitTemplate.convertAndSend(RabbitMqConstant.SIMPLE_QUEUE_NAME, msg);
            }
        }
    }

此生产者发送了 5 条消息

常量类：

public interface RabbitMqConstant {
        
    	// 队列名称
        String SIMPLE_QUEUE_NAME = "simple_queue";
    }

接口调用：

@RestController
    @RequestMapping("/simple")
    public class SimpleController {
        
    
        @Autowired
        private SimpleProducer simpleProducer;
    
        @RequestMapping("/sendMsg")
        public String sendMsg() {
        
            simpleProducer.sendMsg();
            return "SIMPLE-QUEUE";
        }
    }

调用此接口后，会向 MQ 发送 5 条消息，在 MQ 中查看如下：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAc2NvNTI4Mg_size_20_color_FFFFFF_t_70_g_se_x_16]  
启动消费者后，会消费消息

注意：消费者和生产者一定要消费相同的队列

### 3. 在生产者端新建队列 ###

当我们在 MQ 中删除这个 simple\_queue 队列后，再次通过生产者发送消息，发现：虽然队列已不存在，但往这个队列中发送消息时也没有报错。

这就有一个问题：当你发送一条消息时，不管这个队列是否存在，这条消息都不会报错，那我们就会默认它发送成功，如果，这个队列不存在，那么就会存在误判的问题了。

所以，为了避免这种情况，我们在启动生产者时，就去声明一个队列：如果它不存在就创建；存在，就不进行处理。

> 那么，又该如何做呢？

很简单，在生产者端中添加一个配置类即可：

@Configuration
    public class SimpleQueueConfig {
        
    
        // 创建队列
        @Bean
        public Queue simpleQueue() {
        
            return new Queue(RabbitMqConstant.SIMPLE_QUEUE_NAME);
        }
    }

启动生产者端，调用接口，如果 simple\_queue 队列不存在，那么就会创建了。

### 4. 使用 @RabbitListener 注解消费消息 ###

注解 @RabbitListener 既可以标记在类上，也可以标记在方法上：

*  标记在类上：需配合 @RabbitHandler 注解一起使用。当有收到消息的时候，就交给 @RabbitHandler 的方法处理，具体使用哪个方法处理，根据 MessageConverter 转换后的参数类型
 *  标记在方法上：就由指定的方法进行处理

#### 4.1 标记在类上 ####

看看生产者发送消息的代码：

public void sendMsg() {
        
        for (int i = 0; i < 5; i++) {
        
            String msg = "简单模式下的消息 " + i;
            rabbitTemplate.convertAndSend(RabbitMqConstant.SIMPLE_QUEUE_NAME, msg);
        }
    }

往此队列中发送的是 String 类型的数据

看看消费者接收消息的代码：

@RabbitHandler
    public void receive(String message) {
        
        System.out.println("消费者接收到的消息是：" + message);
    }

方法的入参也是 String 类型的。它们类型相匹配，所以，可以消费消息。

现在，我们修改下生产者发送消息的数据类型：

public void sendMsg() {
        
        Map<String, Object> map = new HashMap<>();
        for (int i = 0; i < 5; i++) {
        
            String msg = "简单模式下的消息 " + i;
            map.put(String.valueOf(i), msg);
            rabbitTemplate.convertAndSend(RabbitMqConstant.SIMPLE_QUEUE_NAME, map);
        }
    }

从上面可以看出，我们往队列中发送消息的数据类型为 HashMap。

然后，我们使用消费者接收消息，发现会报错：  
![在这里插入图片描述][be9f66d6e3944c58a062c3713028294b.png]  
数据类型不匹配。

然后，我们再往消费者中添加如下代码：

@RabbitHandler
    public void receive(Map<String, Object> message) {
        
        System.out.println("消费者接收到的消息是：" + message);
    }

这样，就消费消息成功了。

#### 4.2 标记在方法上 ####

生产者代码不变，消费者代码修改如下：

@Component
    public class SimpleConsumer2 {
        
    
        @RabbitListener(queues = {
        "simple_queue"})
        public void receive(String message) {
        
            System.out.println("消费者接收到的消息是：" + message);
        }
    }

只添加 `@RabbitListener` 注解。注意：`simple_queue` 队列必须存在。

### 5. 手动确认消息 ###

修改 application.yml 配置：

server:
      port: 8080
    spring:
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest
        password: guest
        virtual-host: /
        listener:
          simple:
            acknowledge-mode: manual #手动应答
            concurrency: 1 # 最小消费者数量
            max-concurrency: 1 # 最多消费者数量
            retry:
              enabled: true # 是否支持重试

消费者代码修改如下：

@Component
    public class SimpleConsumer2 {
        
    
        @RabbitListener(queues = {
        "simple_queue"})
        public void receive(Message message, Channel channel) throws Exception{
        
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("我是消费信息："+new String(message.getBody()));
        }
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAc2NvNTI4Mg_size_11_color_FFFFFF_t_70_g_se_x_16_pic_left]: https://img-blog.csdnimg.cn/3f96d979ab6842f8a233bcba2436a0e4.jpg?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAc2NvNTI4Mg==,size_11,color_FFFFFF,t_70,g_se,x_16#pic_left
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAc2NvNTI4Mg_size_13_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/ee877cdc706945eb8fc2ab3239ceb793.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAc2NvNTI4Mg==,size_13,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAc2NvNTI4Mg_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/92fba560e20a4227aa2c83213a7f4039.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAc2NvNTI4Mg==,size_20,color_FFFFFF,t_70,g_se,x_16
[be9f66d6e3944c58a062c3713028294b.png]: https://img-blog.csdnimg.cn/be9f66d6e3944c58a062c3713028294b.png