RabbitMQ實現(xiàn)Work Queue工作隊列的示例詳解
RabbitMQ Work Queue工作隊列
工作隊列(又稱任務(wù)隊列)的主要思想是避免立即執(zhí)行資源密集型任務(wù),而不得不等待它完成。
相反我們安排任務(wù)在之后執(zhí)行。我們把任務(wù)封裝為消息并將其發(fā)送到隊列。在后臺運行的工作進(jìn)程將彈出任務(wù)并最終執(zhí)行作業(yè)。當(dāng)有多個工作線程時,這些工作線程將一起處理這些任務(wù)。
多個消費者綁定到一個隊列,同一條消息只會被一個消費者處理。
但是對于工作隊列,可以提高消息的處理速度,避免隊列中的消息堆積。
我們以一個例子來解釋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); } } }
在消費者的服務(wù)模塊中,定義兩個消息監(jiān)聽,分別為listenSimpleQueue1和listenSimpleQueue2,讓它們都監(jiān)聽simple.queue隊列,并且設(shè)置休眠時間,使得消費者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已經(jīng)接收到simple.queue的消息:[" + msg + "]"+ LocalTime.now()); Thread.sleep(20); } @RabbitListener(queues="simple.queue") public void listenSimpleQueue2(String msg) throws InterruptedException { System.err.println("消費者2已經(jīng)接收到simple.queue的消息:[" + msg + "]"+LocalTime.now()); Thread.sleep(200); } }
消費者的application.yaml文件,設(shè)置消費者每次只能獲取一條消息,生產(chǎn)者和消費者的配置文件相似。
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 #每次只能獲取一條消息,處理完成才能獲取下一條消息 控制消費者預(yù)取消息的上限
處理完成后,運行項目,可以得到消費者1和消費者2都能消費消息,并且可以根據(jù)休眠時間有序進(jìn)行工作。
到此這篇關(guān)于RabbitMQ實現(xiàn)Work Queue工作隊列的示例詳解的文章就介紹到這了,更多相關(guān)RabbitMQ Work Queue內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
java jdk1.8 使用stream流進(jìn)行l(wèi)ist 分組歸類操作
這篇文章主要介紹了java jdk1.8 使用stream流進(jìn)行l(wèi)ist 分組歸類操作,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2020-10-10SpringBoot JPA懶加載失效的解決方案(親測有效)
這篇文章主要介紹了SpringBoot JPA懶加載失效的解決方案(親測有效),具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教2021-08-08關(guān)于stream().sorted()以及java中常用的比較器排序
這篇文章主要介紹了關(guān)于stream().sorted()以及java中常用的比較器排序,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-05-05在SpringBoot項目中利用maven的generate插件
今天小編就為大家分享一篇關(guān)于在SpringBoot項目中利用maven的generate插件,小編覺得內(nèi)容挺不錯的,現(xiàn)在分享給大家,具有很好的參考價值,需要的朋友一起跟隨小編來看看吧2019-01-01