RabbitMQ之Work Queue（工作队列）

阿里云国内75折回扣微信号：monov8

阿里云国际，腾讯云国际，低至75折。AWS 93折免费开户实名账号代冲值优惠多多微信号：monov8 飞机：@monov6

前言大家好我是小威24届毕业生曾经在某央企公司实习目前在某税务公司。本篇文章将记录和分享RabbitMQ工作队列相关的知识点。
本篇文章记录的基础知识适合在学Java的小白也适合复习中面试中的大佬🤩🤩。
如果文章有什么需要改进的地方还请大佬不吝赐教👏👏。
小威在此先感谢各位大佬啦~~🤞🤞
在这里插入图片描述

🏠个人主页小威要向诸佬学习呀
🧑个人简介大家好我是小威一个想要与大家共同进步的男人😉😉
目前状况🎉24届毕业生曾经在某央企公司实习目前在某税务公司实习👏👏

💕欢迎大家这里是CSDN我总结知识的地方欢迎来到我的博客我亲爱的大佬😘

在这里插入图片描述

以下正文开始

在这里插入图片描述

工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务而不得不等待它完成。

相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时这些工作线程将一起处理这些任务。
在这里插入图片描述

多个消费者绑定到一个队列同一条消息只会被一个消费者处理。

但是对于工作队列可以提高消息的处理速度避免队列中的消息堆积。

我们以一个例子来解释work queue工作队列。在生产者的服务中添加测试方法通过循环的方式向名为simple.queue队列中发送50条消息代码和详细描述如下

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;
@SpringBootTest
@RunWith(SpringRunner.class)
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testSendMessage2WorkQueue() throws InterruptedException {
        String queueName="simple.queue";//队列名称
        String message = "hello, message_";//发送的消息
        for (int i=1;i<=50;i++){
            rabbitTemplate.convertAndSend(queueName,message+i);
            Thread.sleep(20);
        }
    }
}

在消费者的服务模块中定义两个消息监听分别为listenSimpleQueue1和listenSimpleQueue2让它们都监听simple.queue队列并且设置休眠时间使得消费者1每秒处理50条消息消费者2每秒处理10条消息。

package cn.itcast.mq.listener;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.time.LocalTime;
@Component
public class SpringRabbitListener {
    @RabbitListener(queues="simple.queue")
    public void listenSimpleQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1已经接收到simple.queue的消息:[" + msg +  "]"+ LocalTime.now());
        Thread.sleep(20);
    }
    @RabbitListener(queues="simple.queue")
    public void listenSimpleQueue2(String msg) throws InterruptedException {
        System.err.println("消费者2已经接收到simple.queue的消息:[" + msg +  "]"+LocalTime.now());
        Thread.sleep(200);
    }
}

消费者的application.yaml文件设置消费者每次只能获取一条消息生产者和消费者的配置文件相似。

logging:
  pattern:
    dateformat: MM-dd HH:mm:ss:SSS
spring:
  rabbitmq:
    host: 192.168.220.13*
    port: 5672
    username: user
    password: ******
    virtual-host: /
    Listener:
      simple:
        prefetch: 1 #每次只能获取一条消息处理完成才能获取下一条消息 控制消费者预取消息的上限