消息的可靠投遞

在使用 RabbitMQ 的時候，作為消息發(fā)送方希望杜絕任何消息丟失或者投遞失敗場景。RabbitMQ 為我們提供了兩種方式用來控制消息的投遞可靠性模式。

• confirm 確認模式
• return 退回模式

rabbitmq 整個消息投遞的路徑為：
producer—>rabbitmq broker—>exchange—>queue—>consumer

• 消息從 producer 到 exchange 則會返回一個 confirmCallback 。
• 消息從 exchange–>queue 投遞失敗則會返回一個 returnCallback 。

我們將利用這兩個 callback 控制消息的可靠性投遞

• 設(shè)置ConnectionFactory的publisher-confirms=“true” 開啟 確認模式。

• 使用rabbitTemplate.setConfirmCallback設(shè)置回調(diào)函數(shù)。當消息發(fā)送到exchange后回調(diào)confirm方法。在方法中判斷ack，如果為true，則發(fā)送成功，如果為false，則發(fā)送失敗，需要處理。

• 設(shè)置ConnectionFactory的publisher-returns=“true” 開啟 退回模式。

• 使用rabbitTemplate.setReturnCallback設(shè)置退回函數(shù)，當消息從exchange路由到queue失敗后，如果設(shè)置了rabbitTemplate.setMandatory(true)參數(shù)，則會將消息退回給producer。并執(zhí)行回調(diào)函數(shù)returnedMessage。

• 在RabbitMQ中也提供了事務(wù)機制，但是性能較差

使用channel下列方法，完成事務(wù)控制：

• txSelect(), 用于將當前channel設(shè)置成transaction模式
• txCommit()，用于提交事務(wù)
• txRollback(),用于回滾事務(wù)

Consumer Ack

ack指Acknowledge，確認。 表示消費端收到消息后的確認方式。
有三種確認方式：

• 自動確認：acknowledge=“none”
• 手動確認：acknowledge=“manual”
• 根據(jù)異常情況確認：acknowledge=“auto”

其中自動確認是指，當消息一旦被Consumer接收到，則自動確認收到，并將相應(yīng) message 從 RabbitMQ 的消息緩存中移除。

但是在實際業(yè)務(wù)處理中，很可能消息接收到，業(yè)務(wù)處理出現(xiàn)異常，那么該消息就會丟失

如果設(shè)置了手動確認方式，則需要在業(yè)務(wù)處理成功后，調(diào)用channel.basicAck()，手動簽收

如果出現(xiàn)異常，則調(diào)用channel.basicNack()方法，讓其自動重新發(fā)送消息

• 在rabbit:listener-container標簽中設(shè)置acknowledge屬性，設(shè)置ack方式 none：自動確認，manual：手動確認
• 如果在消費端沒有出現(xiàn)異常，則調(diào)用channel.basicAck(deliveryTag,false);方法確認簽收消息
• 如果出現(xiàn)異常，則在catch中調(diào)用 basicNack或 basicReject，拒絕消息，讓MQ重新發(fā)送消息。

消費端限流

• 在rabbit:listener-container 中配置 prefetch屬性設(shè)置消費端一次拉取多少消息
• 消費端的確認模式一定為手動確認。acknowledge=“manual”

TTL

TTL 全稱 Time To Live（存活時間/過期時間）。當消息到達存活時間后，還沒有被消費，會被自動清除。RabbitMQ可以對消息設(shè)置過期時間，也可以對整個隊列（Queue）設(shè)置過期時間。

• 設(shè)置隊列過期時間使用參數(shù)：x-message-ttl，單位：ms(毫秒)，會對整個隊列消息統(tǒng)一過期。
• 設(shè)置消息過期時間使用參數(shù)：expiration。單位：ms(毫秒)，當該消息在隊列頭部時（消費時），會單獨判斷這一消息是否過期。
• 如果兩者都進行了設(shè)置，以時間短的為準。

死信隊列

死信隊列，英文縮寫：DLX 。Dead Letter Exchange（死信交換機），當消息成為Dead message后，可以被重新發(fā)送到另一個交換機，這個交換機就是DLX

消息成為死信的三種情況：
1. 隊列消息長度到達限制；
2. 消費者拒接消費消息，basicNack/basicReject,并且不把消息重新放入原目標隊列,requeue=false；
3. 原隊列存在消息過期設(shè)置，消息到達超時時間未被消費；

隊列綁定死信交換機：
給隊列設(shè)置參數(shù)： x-dead-letter-exchange 死信交換機名稱和 x-dead-letter-routing-key 發(fā)送給死信交換機的routingkey

1. 死信交換機和死信隊列和普通的沒有區(qū)別
2. 當消息成為死信后，如果該隊列綁定了死信交換機，則消息會被死信交換機重新路由到死信隊列

延遲隊列

延遲隊列，即消息進入隊列后不會立即被消費，只有到達指定時間后，才會被消費。

很可惜，在RabbitMQ中并未提供延遲隊列功能。
但是可以使用：TTL+死信隊列 組合實現(xiàn)延遲隊列的效果。

延遲隊列 指消息進入隊列后，可以被延遲一定時間，再進行消費。

RabbitMQ沒有提供延遲隊列功能，但是可以使用 ： TTL + DLX 來實現(xiàn)延遲隊列效果

日志與監(jiān)控

RabbitMQ日志

RabbitMQ默認日志存放路徑： /var/log/rabbitmq/rabbit@xxx.log
日志包含了RabbitMQ的版本號、Erlang的版本號、RabbitMQ服務(wù)節(jié)點名稱、cookie的hash值、RabbitMQ配置文件地址、內(nèi)存限制、磁盤限制、默認賬戶guest的創(chuàng)建以及權(quán)限配置等等。

web管控臺監(jiān)控

rabbitmqctl管理和監(jiān)控查看隊列# rabbitmqctl list_queues查看exchanges# rabbitmqctl list_exchanges查看用戶# rabbitmqctl list_users查看連接# rabbitmqctl list_connections查看消費者信息# rabbitmqctl list_consumers查看環(huán)境變量# rabbitmqctl environment查看未被確認的隊列# rabbitmqctl list_queues name messages_unacknowledged查看單個隊列的內(nèi)存使用# rabbitmqctl list_queues name memory查看準備就緒的隊列# rabbitmqctl list_queues name messages_ready

消息追蹤

在使用任何消息中間件的過程中，難免會出現(xiàn)某條消息異常丟失的情況。對于RabbitMQ而言，可能是因為生產(chǎn)者或消費者與RabbitMQ斷開了連接，而它們與RabbitMQ又采用了不同的確認機制；也有可能是因為交換器與隊列之間不同的轉(zhuǎn)發(fā)策略；甚至是交換器并沒有與任何隊列進行綁定，生產(chǎn)者又不感知或者沒有采取相應(yīng)的措施；另外RabbitMQ本身的集群策略也可能導(dǎo)致消息的丟失。這個時候就需要有一個較好的機制跟蹤記錄消息的投遞過程，以此協(xié)助開發(fā)和運維人員進行問題的定位。

在RabbitMQ中可以使用Firehose和rabbitmq_tracing插件功能來實現(xiàn)消息追蹤。

消息追蹤-Firehose

firehose的機制是將生產(chǎn)者投遞給rabbitmq的消息，rabbitmq投遞給消費者的消息按照指定的格式發(fā)送到默認的exchange上。這個默認的exchange的名稱為amq.rabbitmq.trace，它是一個topic類型的exchange。發(fā)送到這個exchange上的消息的routing key為 publish.exchangename 和deliver.queuename。其中exchangename和queuename為實際exchange和queue的名稱，分別對應(yīng)生產(chǎn)者投遞到exchange的消息，和消費者從queue上獲取的消息。

注意：打開 trace 會影響消息寫入功能，適當打開后請關(guān)閉。
rabbitmqctl trace_on：開啟Firehose命令
rabbitmqctl trace_off：關(guān)閉Firehose命令

消息追蹤-rabbitmq_tracing

rabbitmq_tracing和Firehose在實現(xiàn)上如出一轍，只不過rabbitmq_tracing的方式比Firehose多了一層GUI的包裝，更容易使用和管理。

啟用插件：rabbitmq-plugins enable rabbitmq_tracing

RabbitMQ 應(yīng)用問題

消息可靠性保障

100%確保消息發(fā)送成功
消息補償

消息冪等性保障

冪等性指一次和多次請求某一個資源，對于資源本身應(yīng)該具有同樣的結(jié)果。也就是說，其任
意多次執(zhí)行對資源本身所產(chǎn)生的影響均與一次執(zhí)行的影響相同。
在MQ中指，消費多條相同的消息，得到與消費該消息一次相同的結(jié)果。

總結(jié)

以上是生活随笔為你收集整理的【消息中间件】RabbitMQ 高级特性与应用问题的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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