在上篇《RabbitMQ-高效的Work模式》中，我們了解了Work模型，該模型包括一個(gè)生產(chǎn)者，一個(gè)消息隊(duì)列和多個(gè)消費(fèi)者。
我們已經(jīng)通過實(shí)例看出消息隊(duì)列中的消息是如何被一個(gè)或者多個(gè)消費(fèi)者消費(fèi)的了，但是對(duì)于具體的實(shí)現(xiàn)細(xì)節(jié)和原理并沒有介紹。這篇就來詳細(xì)介紹下在消息派發(fā)這個(gè)過程中還有那些我們需要關(guān)注的點(diǎn)和細(xì)節(jié)。
這篇主要討論細(xì)節(jié)都集中在接收端，我們還是來看下上篇中，接收端的代碼實(shí)現(xiàn)

package com.ximalaya.openapi.rabbitmq.work;import com.rabbitmq.client.*;import java.io.IOException; /*** Created by jackie on 17/8/4.*/ public class Worker {private static final String TASK_QUEUE_NAME = "task_queue";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("192.168.3.161");final Connection connection = factory.newConnection();final Channel channel = connection.createChannel();channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);System.out.println(" [*] Waiting for messages. To exit press CTRL+C");channel.basicQos(1);final Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println(" [x] Received '" + message + "'");try {doWork(message);} finally {System.out.println(" [x] Done");channel.basicAck(envelope.getDeliveryTag(), false);}}};channel.basicConsume(TASK_QUEUE_NAME, false, consumer);}private static void doWork(String task) {for (char ch : task.toCharArray()) {if (ch == '.') {try {Thread.sleep(2000);} catch (InterruptedException _ignored) {Thread.currentThread().interrupt();}}}} }

消息是怎么合理的派發(fā)給各個(gè)消費(fèi)者的

在上篇介紹的實(shí)例中，我們看到運(yùn)行兩個(gè)消費(fèi)者，這時(shí)候生產(chǎn)者生產(chǎn)的4條信息是均勻的派發(fā)給了兩個(gè)消費(fèi)者。

你可能會(huì)好奇，這個(gè)消息隊(duì)列Queue怎么會(huì)這么“智能”，能夠做到如此公平的進(jìn)行消息派發(fā)。看完下面的場(chǎng)景你可能就不覺得RabbitMQ這樣做是聰明了。

其實(shí)，默認(rèn)情況下的RabbitMQ就是這么“智能”，公平、公正、公開的將4個(gè)消息依次派發(fā)給兩個(gè)消費(fèi)者。如果啟動(dòng)了四個(gè)消費(fèi)者，那也將是每個(gè)消費(fèi)者消費(fèi)一條消息。

這是為什么呢？
RabbitMQ派發(fā)消息默認(rèn)采用的是輪詢機(jī)制，輪詢，顧名思義就是挨個(gè)的派發(fā)，就是第一個(gè)派發(fā)給C1，第二個(gè)派發(fā)給C2，第三個(gè)派發(fā)給C1，第四個(gè)派發(fā)給C4。正常情況下，這樣很好，但是如果遇到某個(gè)消費(fèi)者在消費(fèi)某個(gè)消息時(shí)花費(fèi)時(shí)間很長(zhǎng)或者因?yàn)樽陨碓蚧蛘呔W(wǎng)絡(luò)原因阻塞，那么按照這種輪詢的策略就顯得不合適了。

假設(shè)C2在執(zhí)行第二個(gè)派發(fā)的消息一直卡住，這時(shí)候即使派發(fā)新的消息，C2也無法正常消費(fèi)，如果一直這么盲目的派發(fā)消息給C2，只會(huì)讓更多的消息無法正常消費(fèi)，直至消息隊(duì)列卡住崩潰。

這時(shí)候我們采用一種新的機(jī)制，姑且稱為"公平機(jī)制"。該機(jī)制下，我們?cè)谕粫r(shí)間內(nèi)只給消費(fèi)者派發(fā)一個(gè)消息（派發(fā)的數(shù)量可以人工配置），RabbitMQ只有等到該消費(fèi)者確認(rèn)消費(fèi)了上一條消息后，才會(huì)繼續(xù)派發(fā)下一條消息。

這個(gè)代碼實(shí)現(xiàn)也很簡(jiǎn)單，就是上面接收端中的
channel.basicQos(1);
這里的數(shù)字1就是剛剛提到可以人工配置的派發(fā)消息的數(shù)量。

實(shí)例驗(yàn)證
要驗(yàn)證有basicQos和沒有basicQos，我們需要做一些分析，并對(duì)代碼做部分改動(dòng)。

當(dāng)前啟動(dòng)消費(fèi)端，每個(gè)消費(fèi)者消費(fèi)的時(shí)間都是固定2秒，即使加上basicQos，因?yàn)閮蓚€(gè)隊(duì)列的消費(fèi)速率相同，所以最終還是會(huì)出現(xiàn)兩個(gè)消費(fèi)者各自消費(fèi)兩條消息的情況。

為了營(yíng)造其中一個(gè)消費(fèi)者卡住的情況，我們將后面啟動(dòng)的消費(fèi)者的消費(fèi)時(shí)間設(shè)定為8秒，這樣第一個(gè)消費(fèi)者即使消費(fèi)了三條消息，這時(shí)候第二個(gè)消費(fèi)者仍然卡住，便能看到效果。

下面先看沒有添加basicQos的情況，第一個(gè)和第二個(gè)消費(fèi)者的消費(fèi)時(shí)間分別是2秒和8秒

第一個(gè)Work消費(fèi)時(shí)間是2秒的，第二個(gè)是8秒

看完整個(gè)消費(fèi)過程，會(huì)發(fā)現(xiàn)沒有basiQos設(shè)置，會(huì)執(zhí)行輪詢策略，每個(gè)消費(fèi)者都消費(fèi)了兩個(gè)消息

再看添加basicQos的情況，第一個(gè)和第二個(gè)消費(fèi)者的消費(fèi)時(shí)間同樣分別是2秒和8秒

同樣，第一個(gè)Work消費(fèi)時(shí)間是2秒的，第二個(gè)是8秒

看完整個(gè)消費(fèi)過程，會(huì)發(fā)現(xiàn)有basiQos設(shè)置，會(huì)執(zhí)行公平機(jī)制，第一個(gè)消息給C1，第二個(gè)給C2，第三個(gè)消息來的時(shí)候，這時(shí)候發(fā)現(xiàn)C2還在消費(fèi)，就派發(fā)給了已經(jīng)消費(fèi)完空閑的C1，第四個(gè)消息來的時(shí)候，發(fā)現(xiàn)C2仍然在消費(fèi)，這時(shí)候就把消息派發(fā)給了消費(fèi)完第三個(gè)消息的C1，C1總共消費(fèi)3條消息用時(shí)6秒，而C2消費(fèi)一條消息時(shí)8秒，所以這就是公平機(jī)制。

對(duì)比完后，我們發(fā)現(xiàn)這種公平機(jī)制更加合理，能夠很好的做到負(fù)載均衡，避免因?yàn)椴活櫹M(fèi)者的消費(fèi)情況而盲目派發(fā)情況的出現(xiàn)。

如何保證派發(fā)出去的消息不丟失

現(xiàn)在如果出現(xiàn)這樣的一種情況：消息從Queue中取出，但是沒有消費(fèi)者因?yàn)楦鞣N情況并沒有完成這條消息的消費(fèi)，但是這條消息已經(jīng)從內(nèi)存中刪除了，這就意味著這個(gè)消息模型就失去了這條消息，這種意外在大多數(shù)場(chǎng)景下是不允許出現(xiàn)的。

為什么會(huì)出現(xiàn)這種情況呢？
因?yàn)橄⒊鋈サ臅r(shí)候，RabbitMQ就將其從Queue中刪除，也就是從內(nèi)存中刪除，這樣做的假設(shè)前提就是默認(rèn)為這條消息能夠被正常消費(fèi)掉，但事實(shí)情況往往并非如此。如果此時(shí)我們加上一個(gè)確認(rèn)機(jī)制，類似于TCP的三次握手，問題就能夠得到解決。

RabbitMQ將消息派發(fā)出去后并不立馬將消息從內(nèi)存中刪除，等到消費(fèi)端完成消費(fèi)返回一個(gè)ack的標(biāo)識(shí)，RabbitMQ接收到這個(gè)字段后認(rèn)為消息時(shí)正常消費(fèi)了在完成刪除。如果沒有收到確認(rèn)標(biāo)識(shí)ack，則認(rèn)為消息違背正常消費(fèi)，則會(huì)重新取回該條，采用輪詢或者其他機(jī)制將其派發(fā)到下一個(gè)消費(fèi)者供其消費(fèi)。

實(shí)例
在接收端將代碼中
channel.basicConsume(TASK_QUEUE_NAME, false, consumer);
將basicConsume函數(shù)的第二個(gè)參數(shù)改為true標(biāo)識(shí)autoAck=true，即自動(dòng)確認(rèn)，如果設(shè)置為false，則表示需要接收端手工確認(rèn)。

上篇，我們用的就是false的情況，即手動(dòng)確認(rèn)方式，所以在上篇的運(yùn)行接口我們看到Unacknowleged標(biāo)識(shí)一直從1變?yōu)?，是說明采用的是一條一條確認(rèn)的機(jī)制，從第一條消息一直到第四條消息消費(fèi)完成。

下面我們看看autoAck=true運(yùn)行時(shí)Ready和Unacknowleged指標(biāo)的變化趨勢(shì)，我們只啟動(dòng)一個(gè)消費(fèi)者

請(qǐng)點(diǎn)擊此處輸入圖片描述
從運(yùn)行過程可以發(fā)現(xiàn)，Unackowleged從0->4->3->2->1->0，autoAck=false是為0->1->1->1->1->0
說明autoAck=false時(shí)是一次性派發(fā)了4條信息，沒有顧忌消費(fèi)者是否有發(fā)送確認(rèn)標(biāo)識(shí)。之后消費(fèi)者再依次完成消費(fèi)。
?如果您覺得閱讀本文對(duì)您有幫助，請(qǐng)點(diǎn)一下“推薦”按鈕，您的“推薦”將是我最大的寫作動(dòng)力！如果您想持續(xù)關(guān)注我的文章，請(qǐng)掃描二維碼，關(guān)注JackieZheng的微信公眾號(hào)，我會(huì)將我的文章推送給您，并和您一起分享我日常閱讀過的優(yōu)質(zhì)文章。

轉(zhuǎn)載于:https://www.cnblogs.com/bigdataZJ/p/rabbitmq4.html

總結(jié)

以上是生活随笔為你收集整理的RabbitMQ入门-消息派发那些事儿的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

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