一、簡介

是一個(gè)性能也許不是很高，但是是一個(gè)可靠的消息傳遞機(jī)制進(jìn)行與平臺(tái)無關(guān)的數(shù)據(jù)交流，并基于數(shù)據(jù)通信來進(jìn)行分布式系統(tǒng)的集成。

1、作用與特性

• 系統(tǒng)解耦

• 冗余（存儲(chǔ)數(shù)據(jù)）

• 可擴(kuò)展集群部署

• 可用性鏡像隊(duì)列

• 削峰

• 可恢復(fù)性

• 緩沖

• 異步通信

• 可靠性生產(chǎn)者消息確認(rèn)消費(fèi)者消息確認(rèn)相關(guān)組件支持持久化

• 靈活的路由

• 支持多種協(xié)議

• 支持多種語言客戶端

• 支持友好管理界面

• 支持插件機(jī)制

2、相關(guān)組件介紹

Rabbit是AMQP協(xié)議的一個(gè)實(shí)現(xiàn)，所以在熟悉了AMQP協(xié)議之后可以更好地使用Rabbit

• 生產(chǎn)者與消費(fèi)者

• 隊(duì)列：用于存儲(chǔ)消息

• 交換機(jī)：根據(jù)綁定鍵與路由鍵將消息投遞到隊(duì)列主題 topic通過“.”分割字符串“*”匹配一個(gè)單詞“#”匹配一個(gè)或多個(gè)單詞留意這里跟正則表達(dá)式有些許區(qū)別直接 direct路由鍵完全匹配綁定鍵彈出 fanout頭 header

• 綁定：將交換機(jī)跟隊(duì)列，或?qū)⒔粨Q機(jī)跟交換機(jī)通過某種規(guī)則綁定起來綁定鍵：投遞的某種規(guī)則，根據(jù)交換機(jī)類型所覺得的匹配規(guī)則路由鍵：投遞消息時(shí)代表這條消息的投遞規(guī)則，由交換機(jī)決定是否擊中綁定鍵而進(jìn)行投遞

• 連接：TCP協(xié)議建立的一條可靠連接

• 信道：建立在鏈接上面的虛擬鏈接

3、運(yùn)轉(zhuǎn)流程

以下流程指的是比較通用的業(yè)務(wù)場景

• 生產(chǎn)者側(cè)鏈接消息隊(duì)列并建立信道聲明一個(gè)交換機(jī)聲明一個(gè)隊(duì)列綁定交換機(jī)與隊(duì)列之間的關(guān)系指定路由鍵并發(fā)送消息關(guān)閉信道關(guān)閉連接

• 消費(fèi)者側(cè)鏈接消息隊(duì)列并建立信道聲明一個(gè)交換機(jī)聲明一個(gè)隊(duì)列綁定交換機(jī)與隊(duì)列之間的關(guān)系訂閱某條隊(duì)列并接收消息消息確認(rèn)關(guān)閉信道關(guān)閉連接

二、開發(fā)要點(diǎn)

1、連接RabbitMQ

信道不能在線程間共享，可能會(huì)出現(xiàn)錯(cuò)誤的通信幀交錯(cuò)，同時(shí)也會(huì)影響發(fā)送確認(rèn)的問題

2、交互動(dòng)作

2.1、交換機(jī)的描述exchangeDeclare

相關(guān)的屬性

• exchange：交換器名稱

• type：類型

• durable：是否持久化

• autoDeleted：是否自動(dòng)刪除，至少有一個(gè)隊(duì)列或交換器與該交換器綁定，之后所有與這個(gè)交換器綁定的隊(duì)列或交換器與此解綁。

• inernal：是否內(nèi)置。內(nèi)置交換機(jī)只能通過交換路由到交換器這種方式使用

• argument：其他一些結(jié)構(gòu)化參數(shù)

類似的方法

• exchangeDeclarePassive：描述的交換器必須存在，否則拋出異常

• exchangeDeclareNoWait：不需要服務(wù)器返回OK，建議不用，防止創(chuàng)建失敗

注意事項(xiàng)

• 生產(chǎn)者跟消費(fèi)者都可以聲明一個(gè)交換器與隊(duì)列，如果嘗試聲明一個(gè)已經(jīng)存在的交換器或隊(duì)列，如果參數(shù)完全匹配，那么不做任何事情。如果發(fā)現(xiàn)不匹配，出現(xiàn)沖突則響應(yīng)一個(gè)異常。

2.2、隊(duì)列的描述queueDeclared

相關(guān)屬性

• queue：隊(duì)列的名稱

• durable：是否持久化

• exclusive：設(shè)置是否排他，排他的意思是只對當(dāng)前連接可見。

• autoDeleted：是否刪除。至少有一個(gè)消費(fèi)者連接到這個(gè)隊(duì)列，之后所有與這個(gè)隊(duì)列的消費(fèi)者都斷開時(shí)，才會(huì)自動(dòng)刪除。因?yàn)樯a(chǎn)者創(chuàng)建了這個(gè)隊(duì)列，或者沒有消費(fèi)者客戶端與這個(gè)隊(duì)列連接時(shí)，都不會(huì)自動(dòng)刪除這個(gè)隊(duì)列

• arguments：隊(duì)列參數(shù)

類似方法

• queueDeclareNoWait：不需要服務(wù)器返回OK，建議不使用

• queueDeclarePassive：描述的隊(duì)列必須存在，否則響應(yīng)異常

2.3、隊(duì)列綁定queueBind

相關(guān)屬性

• queue：隊(duì)列名稱

• exchange：交換機(jī)名稱

• routingKey：綁定鍵

• argument：綁定參數(shù)

2.4、交換機(jī)綁定exchangeBind

指的是將交換機(jī)與交換機(jī)綁定到一起，使用方法跟隊(duì)列綁定相似

2.5、發(fā)布消息basicPublish

相關(guān)參數(shù)

• exchange：交換器名稱

• routingKey：路由鍵

• props：基本屬性集

• mandatory：確認(rèn)是否能投遞到隊(duì)列中，如果不是則返回

• immediate：是否立即投遞到消費(fèi)者手里，如果不是則返回

2.6、推模式訂閱消息basicConsume

相關(guān)屬性

• queue：隊(duì)列名稱

• autoAck：設(shè)置是否自動(dòng)確認(rèn)

• consumerTag：設(shè)置消費(fèi)者標(biāo)簽

• noLocal：設(shè)置為true表示不能將同一個(gè)鏈接中生產(chǎn)者發(fā)送的消息傳送給這個(gè)鏈接中的消費(fèi)者

• exclusive：設(shè)置是否排他

• argument：其他參數(shù)

• callback：消費(fèi)者回調(diào)函數(shù)，用來處理消息

注意事項(xiàng)：會(huì)收到Basic Qos的限制

2.7、拉模式訂閱消息basicGet

建議不要使用while+basicGet會(huì)嚴(yán)重影響性能

2.8、消費(fèi)端的確認(rèn)與拒絕

• 訂閱隊(duì)列時(shí)指定autoAck參數(shù)true：自動(dòng)確認(rèn)false：等待消費(fèi)者確認(rèn)

• 拒絕一條：basicReject

• 拒絕之前所有：basicNack

• 請求重新發(fā)送還未確認(rèn)的消息：basicRecover

三、高階特性

1、mandatory

確認(rèn)消息必須投遞到隊(duì)列中，如果無法匹配則返回生產(chǎn)者

2、immediate

• 確認(rèn)消息必須投遞到消費(fèi)者手中，該方法已過期

• 使用死信隊(duì)列加生存時(shí)間代替處理此問題

3、備份交換器AE

未被路由到的消息可以投遞到該交換機(jī)

配置方法：在描述交換機(jī)的時(shí)候添加“altermate-exchange”指定一個(gè)備份交換機(jī)。

特殊情況

• 如果不存在備份交換機(jī)，那么消息會(huì)丟失

• 如果備份交換機(jī)，沒有綁定任何隊(duì)列，那么消息會(huì)丟失

• 如果備份交換機(jī)跟mandatory參數(shù)一起使用，那么mandatory參數(shù)無效

4、過期時(shí)間（TTL）

指的是消息超過這個(gè)存活時(shí)間將會(huì)被消息隊(duì)列剔除

• 通過消息參數(shù)設(shè)置“expiration”

• 通過隊(duì)列參數(shù)設(shè)置“x-message-ttl”

如果是通過隊(duì)列的屬性設(shè)置，會(huì)從隊(duì)列頭固定掃描剔除過期消息。如果通過消息屬性設(shè)置，那么會(huì)在被消費(fèi)的時(shí)候剔除

5、死信隊(duì)列（DLX）

當(dāng)消息變成死信的時(shí)候，將會(huì)被投遞到該交換器中

消息變成死信的特征

• 消息被拒絕

• 消息過期

• 隊(duì)列達(dá)到最大長度

設(shè)置方法：在描述隊(duì)列時(shí)，加入x-dead-letter-exchange參數(shù)指定一條死信隊(duì)列

6、延遲隊(duì)列

rabbit沒有這種隊(duì)列的默認(rèn)實(shí)現(xiàn)，但是可以通過死信隊(duì)列+TTL來完成模擬

7、優(yōu)先級(jí)隊(duì)列

優(yōu)先級(jí)高的消息具備優(yōu)先被消費(fèi)的特權(quán)

設(shè)置方法：通過隊(duì)列的x-max-priority參數(shù)設(shè)置

注意事項(xiàng)：如果消息者消費(fèi)的速度太快則體現(xiàn)不了該能力

8、RPC方案支持

方案流程：

• 調(diào)用方把參數(shù)放置進(jìn)一個(gè)RPC隊(duì)列，并制定回調(diào)隊(duì)列

• 被調(diào)用方訂閱RPC隊(duì)列，進(jìn)行計(jì)算后把結(jié)果投遞到調(diào)用方指定的回調(diào)隊(duì)列

• 調(diào)用方從調(diào)用隊(duì)列里面獲取響應(yīng)內(nèi)容

9、持久化

• 交換機(jī)持久化

• 隊(duì)列持久化

• 消息持久化

注意事項(xiàng)

• 三個(gè)組件都需要設(shè)置持久化屬性，那么持久化才能起到效果

• 持久化會(huì)嚴(yán)重影響性能

• 落盤不是實(shí)時(shí)的

• 并不能保證100%的可靠性

10、生產(chǎn)者確認(rèn)機(jī)制

• 事務(wù)相關(guān)命令channel.txSelectchannel.txCommitchannel.txRollback執(zhí)行出錯(cuò)，捕捉異常，隊(duì)列回滾非常消耗性能

• 發(fā)送方確認(rèn)設(shè)置方法：channel.confirmSelect確認(rèn)方式：單條確認(rèn)：channel.waitForConfirms批量確認(rèn)：即等待到一定數(shù)量的消息之后再執(zhí)行確認(rèn)動(dòng)作。批量重傳會(huì)影響性能。異步確認(rèn)（建議）：通過ConfirmListener來異步監(jiān)聽，通過SortedSet來存儲(chǔ)那些未被確認(rèn)的消息id。編程會(huì)比較復(fù)雜

• 注意這個(gè)機(jī)制跟mandatory存在區(qū)別，生產(chǎn)者確認(rèn)機(jī)制是確保消息可以到達(dá)rabbitmq，而不保證能到達(dá)隊(duì)列。且確認(rèn)機(jī)制跟事務(wù)機(jī)制是互斥的，不能同時(shí)使用

11、消費(fèi)端要點(diǎn)

• 消息分發(fā)同隊(duì)列輪詢分發(fā)basicQos控制channal最多接收的消息數(shù)，使用gobal控制整個(gè)信道的消息數(shù)

• 消息順序性保證在多種場景下均可能發(fā)生消息順序不一致的場景可以使用序列id對消息進(jìn)行標(biāo)志，使用緩存等中間件進(jìn)行重排序

• 消息傳輸保障最多發(fā)一次（存在消息丟失的可能性，消息不會(huì)重復(fù)）最少發(fā)一次（存在消息重復(fù)的可能性）發(fā)送確認(rèn)投遞確認(rèn)持久化處理接收處理恰好發(fā)一次
  用

11、消費(fèi)端要點(diǎn)

• 消息分發(fā)同隊(duì)列輪詢分發(fā)basicQos控制channal最多接收的消息數(shù)，使用gobal控制整個(gè)信道的消息數(shù)

• 消息順序性保證在多種場景下均可能發(fā)生消息順序不一致的場景可以使用序列id對消息進(jìn)行標(biāo)志，使用緩存等中間件進(jìn)行重排序

• 消息傳輸保障最多發(fā)一次（存在消息丟失的可能性，消息不會(huì)重復(fù)）最少發(fā)一次（存在消息重復(fù)的可能性）發(fā)送確認(rèn)投遞確認(rèn)持久化處理接收處理恰好發(fā)一次難以實(shí)現(xiàn)