【RabbitMQ】Basic Queue 简单队列模型与WorkQueue

ゝ一纸荒年。 2024-03-27 03:53 187阅读 0赞

**目录**

SpringAMQP

1 Basic Queue 简单队列模型

1.1.消息发送

1.2.消息接收

1.3.测试

2 WorkQueue

2.1.消息发送

2.2.消息接收

2.3.测试

2.4.能者多劳

2.5.总结

## SpringAMQP ##

SpringAMQP是基于RabbitMQ封装的一套模板，并且还利用SpringBoot对其实现了自动装配，使用起来非常方便。

SpringAmqp的官方地址：[Spring AMQP][]

![1d0332c314f14d2cb40970b19d7461e8.png][]

![1b379ef6441e4c00a79a7004edc08a99.png][]

SpringAMQP提供了三个功能：

*  自动声明队列、交换机及其绑定关系
 *  基于注解的监听器模式，异步接收消息
 *  封装了RabbitTemplate工具，用于发送消息

## 1 Basic Queue 简单队列模型 ##

在父工程mq-demo中引入依赖

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>

### 1.1.消息发送 ###

首先配置MQ地址，在publisher服务的application.yml中添加配置：

spring:
      rabbitmq:
        host: 192.168.150.101 # 主机名
        port: 5672 # 端口
        virtual-host: / # 虚拟主机
        username: itcast # 用户名
        password: 123321 # 密码

然后在publisher服务中编写测试类SpringAmqpTest，并利用RabbitTemplate实现消息发送：

package cn.itcast.mq.spring;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class SpringAmqpTest {
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Test
        public void testSimpleQueue() {
            // 队列名称
            String queueName = "simple.queue";
            // 消息
            String message = "hello, spring amqp!";
            // 发送消息
            rabbitTemplate.convertAndSend(queueName, message);
        }
    }

### 1.2.消息接收 ###

首先配置MQ地址，在consumer服务的application.yml中添加配置：

然后在consumer服务的`cn.itcast.mq.listener`包中新建一个类SpringRabbitListener，代码如

下：

package cn.itcast.mq.listener;
    
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class SpringRabbitListener {
    
        @RabbitListener(queues = "simple.queue")
        public void listenSimpleQueueMessage(String msg) throws InterruptedException {
            System.out.println("spring 消费者接收到消息：【" + msg + "】");
        }
    }

### 1.3.测试 ###

启动consumer服务，然后在publisher服务中运行测试代码，发送MQ消息

## 2 WorkQueue ##

Work queues，也被称为（Task queues），任务模型。简单来说就是**让多个消费者绑定到一个队**

**列，共同消费队列中的消息**。

![3147b740653d47ec9e908408b3bac950.png][]

当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。

此时就可以使用work 模型，多个消费者共同处理消息处理，速度就能大大提高了。

### 2.1.消息发送 ###

这次我们循环发送，模拟大量消息堆积现象。

在publisher服务中的SpringAmqpTest类中添加一个测试方法：

/**
         * workQueue
         * 向队列中不停发送消息，模拟消息堆积。
         */
    @Test
    public void testWorkQueue() throws InterruptedException {
        // 队列名称
        String queueName = "simple.queue";
        // 消息
        String message = "hello, message_";
        for (int i = 0; i < 50; i++) {
            // 发送消息
            rabbitTemplate.convertAndSend(queueName, message + i);
            Thread.sleep(20);
        }
    }

### 2.2.消息接收 ###

要模拟多个消费者绑定同一个队列，我们在consumer服务的SpringRabbitListener中添加2个新的

方法：

@RabbitListener(queues = "simple.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
        Thread.sleep(20);
    }
    
    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
        Thread.sleep(200);
    }

注意到这个消费者sleep了1000秒，模拟任务耗时。

### 2.3.测试 ###

启动ConsumerApplication后，在执行publisher服务中刚刚编写的发送测试方法testWorkQueue。

可以看到消费者1很快完成了自己的25条消息。消费者2却在缓慢的处理自己的25条消息。

也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。这样显然是有问题的。

### 2.4.能者多劳 ###

在spring中有一个简单的配置，可以解决这个问题。我们修改consumer服务的application.yml文

件，添加配置：

spring:
      rabbitmq:
        listener:
          simple:
            prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

### 2.5.总结 ###

Work模型的使用：

*  多个消费者绑定到一个队列，同一条消息只会被一个消费者处理
 *  通过设置prefetch来控制消费者预取的消息数量

[Spring AMQP]: https://spring.io/projects/spring-amqp
[1d0332c314f14d2cb40970b19d7461e8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/37bb0b1913f84a5ebcfb58ba09d26f15.png
[1b379ef6441e4c00a79a7004edc08a99.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/3d71a450139145e6a814a49032d51d8a.png
[3147b740653d47ec9e908408b3bac950.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/1b2e4672b5f84705a07e480b3221f728.png