快捷導(dǎo)航

RabbitMQ實現(xiàn)Work Queue工作隊列的示例詳解

更新時間：2023年01月10日 10:28:56 作者：小威要向諸佬學(xué)習(xí)呀

工作隊列(又稱任務(wù)隊列)的主要思想是避免立即執(zhí)行資源密集型任務(wù)，而不得不等待它完成。本篇文章將記錄和分享RabbitMQ工作隊列相關(guān)的知識點(diǎn)，希望對大家有所幫助

RabbitMQ Work Queue工作隊列

工作隊列(又稱任務(wù)隊列)的主要思想是避免立即執(zhí)行資源密集型任務(wù)，而不得不等待它完成。

相反我們安排任務(wù)在之后執(zhí)行。我們把任務(wù)封裝為消息并將其發(fā)送到隊列。在后臺運(yùn)行的工作進(jìn)程將彈出任務(wù)并最終執(zhí)行作業(yè)。當(dāng)有多個工作線程時，這些工作線程將一起處理這些任務(wù)。

多個消費(fèi)者綁定到一個隊列，同一條消息只會被一個消費(fèi)者處理。

但是對于工作隊列，可以提高消息的處理速度，避免隊列中的消息堆積。

我們以一個例子來解釋work queue工作隊列。在生產(chǎn)者的服務(wù)中添加測試方法，通過循環(huán)的方式，向名為simple.queue隊列中發(fā)送50條消息，代碼和詳細(xì)描述如下：

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_";//發(fā)送的消息
        for (int i=1;i<=50;i++){
            rabbitTemplate.convertAndSend(queueName,message+i);
            Thread.sleep(20);
        }
    }
}

在消費(fèi)者的服務(wù)模塊中，定義兩個消息監(jiān)聽，分別為listenSimpleQueue1和listenSimpleQueue2，讓它們都監(jiān)聽simple.queue隊列，并且設(shè)置休眠時間，使得消費(fèi)者1每秒處理50條消息，消費(fèi)者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("消費(fèi)者1已經(jīng)接收到simple.queue的消息:[" + msg +  "]"+ LocalTime.now());
        Thread.sleep(20);
    }
    @RabbitListener(queues="simple.queue")
    public void listenSimpleQueue2(String msg) throws InterruptedException {
        System.err.println("消費(fèi)者2已經(jīng)接收到simple.queue的消息:[" + msg +  "]"+LocalTime.now());
        Thread.sleep(200);
    }
}

消費(fèi)者的application.yaml文件，設(shè)置消費(fèi)者每次只能獲取一條消息，生產(chǎn)者和消費(fèi)者的配置文件相似。

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 #每次只能獲取一條消息，處理完成才能獲取下一條消息 控制消費(fèi)者預(yù)取消息的上限

處理完成后，運(yùn)行項目，可以得到消費(fèi)者1和消費(fèi)者2都能消費(fèi)消息，并且可以根據(jù)休眠時間有序進(jìn)行工作。