前言

RabbitMQ六種隊(duì)列模式-簡單隊(duì)列
RabbitMQ六種隊(duì)列模式-工作隊(duì)列?[本文]
RabbitMQ六種隊(duì)列模式-發(fā)布訂閱
RabbitMQ六種隊(duì)列模式-路由模式
RabbitMQ六種隊(duì)列模式-主題模式

上文我們了解了 RabbitMQ 六種隊(duì)列模式中的簡單隊(duì)列，代碼也是非常的簡單，比較容易理解。

但是簡單隊(duì)列有個(gè)缺點(diǎn)，簡單隊(duì)列是一一對應(yīng)的關(guān)系，即點(diǎn)對點(diǎn)，一個(gè)生產(chǎn)者對應(yīng)一個(gè)消費(fèi)者，按照這個(gè)邏輯，如果我們有一些比較耗時(shí)的任務(wù)，也就意味著需要大量的時(shí)間才能處理完畢，顯然簡單隊(duì)列模式并不能滿足我們的工作需求，我們今天再來看看工作隊(duì)列。

文章目錄

1. 什么是工作隊(duì)列2. 代碼部分2.1 生產(chǎn)者2.2 消費(fèi)者3. 循環(huán)分發(fā)3.1 啟動(dòng)生產(chǎn)者3.2 啟動(dòng)兩個(gè)消費(fèi)者3.3 公平分發(fā)4. 消息持久化4.1 問題背景4.2 參數(shù)配置5. 工作隊(duì)列總結(jié)

1. 什么是工作隊(duì)列

工作隊(duì)列：用來將耗時(shí)的任務(wù)分發(fā)給多個(gè)消費(fèi)者（工作者）

主要解決問題：處理資源密集型任務(wù)，并且還要等他完成。有了工作隊(duì)列，我們就可以將具體的工作放到后面去做，將工作封裝為一個(gè)消息，發(fā)送到隊(duì)列中，一個(gè)工作進(jìn)程就可以取出消息并完成工作。如果啟動(dòng)了多個(gè)工作進(jìn)程，那么工作就可以在多個(gè)進(jìn)程間共享。

工作隊(duì)列也稱為公平性隊(duì)列模式，怎么個(gè)說法呢？

循環(huán)分發(fā)，假如我們擁有兩個(gè)消費(fèi)者，默認(rèn)情況下，RabbitMQ 將按順序?qū)⒚織l消息發(fā)送給下一個(gè)消費(fèi)者，平均而言，每個(gè)消費(fèi)者將獲得相同數(shù)量的消息，這種分發(fā)消息的方式稱為輪詢。

看代碼吧。

2. 代碼部分

2.1 生產(chǎn)者

創(chuàng)建50個(gè)消息

public?class?Producer2?{/**?隊(duì)列名稱?*/private?static?final?String?QUEUE_NAME?=?"test_queue";public?static?void?main(String[]?args)?throws?IOException,?TimeoutException?{/**?1.獲取連接?*/Connection?newConnection?=?MQConnectionUtils.newConnection();/**?2.創(chuàng)建通道?*/Channel?channel?=?newConnection.createChannel();/**3.創(chuàng)建隊(duì)列聲明?*/channel.queueDeclare(QUEUE_NAME,?false,?false,?false,?null);/**保證一次只分發(fā)一次?限制發(fā)送給同一個(gè)消費(fèi)者?不得超過一條消息?*/channel.basicQos(1);for?(int?i?=?1;?i?<=?50;?i++)?{String?msg?=?"生產(chǎn)者消息_"?+?i;System.out.println("生產(chǎn)者發(fā)送消息:"?+?msg);/**4.發(fā)送消息?*/channel.basicPublish("",?QUEUE_NAME,?null,?msg.getBytes());}channel.close();newConnection.close();}}

2.2 消費(fèi)者

public?class?Customer2_1?{/***?隊(duì)列名稱*/private?static?final?String?QUEUE_NAME?=?"test_queue";public?static?void?main(String[]?args)?throws?IOException,?TimeoutException?{System.out.println("001");/**?1.獲取連接?*/Connection?newConnection?=?MQConnectionUtils.newConnection();/**?2.獲取通道?*/final?Channel?channel?=?newConnection.createChannel();channel.queueDeclare(QUEUE_NAME,?false,?false,?false,?null);/**?保證一次只分發(fā)一次?限制發(fā)送給同一個(gè)消費(fèi)者?不得超過一條消息?*/channel.basicQos(1);DefaultConsumer?defaultConsumer?=?new?DefaultConsumer(channel)?{@Overridepublic?void?handleDelivery(String?consumerTag,?Envelope?envelope,?AMQP.BasicProperties?properties,?byte[]?body)throws?IOException?{String?msgString?=?new?String(body,?"UTF-8");System.out.println("消費(fèi)者獲取消息:"?+?msgString);try?{Thread.sleep(1000);}?catch?(Exception?e)?{}?finally?{/**?手動(dòng)回執(zhí)消息?*/channel.basicAck(envelope.getDeliveryTag(),?false);}}};/**?3.監(jiān)聽隊(duì)列?*/channel.basicConsume(QUEUE_NAME,?false,?defaultConsumer);}}

3. 循環(huán)分發(fā)

3.1 啟動(dòng)生產(chǎn)者

3.2 啟動(dòng)兩個(gè)消費(fèi)者

在生產(chǎn)者中我們發(fā)送了50條消息進(jìn)入隊(duì)列，而上方消費(fèi)者啟動(dòng)圖里很明顯的看到輪詢的效果，就是每個(gè)消費(fèi)者會(huì)分到相同的隊(duì)列任務(wù)。

3.3 公平分發(fā)

由于上方模擬的是非常簡單的消息隊(duì)列的消費(fèi)，假如有一些非常耗時(shí)的任務(wù)，某個(gè)消費(fèi)者在緩慢地進(jìn)行處理，而另一個(gè)消費(fèi)者則空閑，顯然是非常消耗資源的。

再舉一個(gè)例子，一個(gè)1年的程序員，跟一個(gè)3年的程序員，分配相同的任務(wù)量，明顯3年的程序員處理起來更加得心應(yīng)手，很快就無所事事了，但是3年的程序員拿著非常高的薪資！顯然3年的程序員應(yīng)該承擔(dān)更多的責(zé)任，那怎么辦呢？

公平分發(fā)。

其實(shí)發(fā)生上述問題的原因是 RabbitMQ 收到消息后就立即分發(fā)出去，而沒有確認(rèn)各個(gè)工作者未返回確認(rèn)的消息數(shù)量，類似于TCP/UDP中的UDP，面向無連接。

因此我們可以使用 basicQos 方法，并將參數(shù) prefetchCount 設(shè)為1，告訴 RabbitMQ 我每次值處理一條消息，你要等我處理完了再分給我下一個(gè)。這樣 RabbitMQ 就不會(huì)輪流分發(fā)了，而是尋找空閑的工作者進(jìn)行分發(fā)。

關(guān)鍵性代碼：

/**?2.獲取通道?*/ final?Channel?channel?=?newConnection.createChannel(); channel.queueDeclare(QUEUE_NAME,?false,?false,?false,?null); /**?保證一次只分發(fā)一次?限制發(fā)送給同一個(gè)消費(fèi)者?不得超過一條消息?*/ channel.basicQos(1);

4. 消息持久化

4.1 問題背景

上邊我們提到的公平分發(fā)是由消費(fèi)者收取消息時(shí)確認(rèn)解決的，但是這里面又會(huì)出現(xiàn)被 kill 的情況。

當(dāng)有多個(gè)消費(fèi)者同時(shí)收取消息，且每個(gè)消費(fèi)者在接收消息的同時(shí)，還要處理其它的事情，且會(huì)消耗很長的時(shí)間。在此過程中可能會(huì)出現(xiàn)一些意外，比如消息接收到一半的時(shí)候，一個(gè)消費(fèi)者死掉了。

這種情況要使用消息接收確認(rèn)機(jī)制，可以執(zhí)行上次宕機(jī)的消費(fèi)者沒有完成的事情。

但是在默認(rèn)情況下，我們程序創(chuàng)建的消息隊(duì)列以及存放在隊(duì)列里面的消息，都是非持久化的。當(dāng)RabbitMQ死掉了或者重啟了，上次創(chuàng)建的隊(duì)列、消息都不會(huì)保存。

怎么辦呢？

4.2 參數(shù)配置

參數(shù)配置一：生產(chǎn)者創(chuàng)建隊(duì)列聲明時(shí)，修改第二個(gè)參數(shù)為 true

/**3.創(chuàng)建隊(duì)列聲明?*/ channel.queueDeclare(QUEUE_NAME,?true,?false,?false,?null);

參數(shù)配置二：生產(chǎn)者發(fā)送消息時(shí)，修改第三個(gè)參數(shù)為MessageProperties.PERSISTENT_TEXT_PLAIN

for?(int?i?=?1;?i?<=?50;?i++)?{String?msg?=?"生產(chǎn)者消息_"?+?i;System.out.println("生產(chǎn)者發(fā)送消息:"?+?msg);/**4.發(fā)送消息?*/channel.basicPublish("",?QUEUE_NAME,?MessageProperties.PERSISTENT_TEXT_PLAIN,?msg.getBytes()); }

5. 工作隊(duì)列總結(jié)

1、循環(huán)分發(fā)：消費(fèi)者端在信道上打開消息應(yīng)答機(jī)制，并確保能返回接收消息的確認(rèn)信息，這樣可以保證消費(fèi)者發(fā)生故障也不會(huì)丟失消息。

2、消息持久化：服務(wù)器端和客戶端都要指定隊(duì)列的持久化和消息的持久化，這樣可以保證RabbitMQ重啟，隊(duì)列和消息也不會(huì)丟失。

3、公平分發(fā)：指定消費(fèi)者接收的消息個(gè)數(shù)，避免出現(xiàn)消息均勻推送出現(xiàn)的資源不合理利用的問題。

案例代碼：https://www.lanzous.com/i5ydu6d

總結(jié)

以上是生活随笔為你收集整理的【转】RabbitMQ六种队列模式-2.工作队列模式的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。