RabbitMQ

• RabbitMQ安裝
• RabbitMQ使用
• • RabbitMQ發(fā)送消息步驟圖
  • 公平消費(fèi)和消息可靠性傳遞
  • 防止重復(fù)消費(fèi)
  • 有序消費(fèi)
  • 消息堆積怎么處理
• spring集成RabbitMQ使用
• RabbitMQ知識(shí)點(diǎn)
• SpringBoot集成RabbitMQ源碼學(xué)習(xí)

RabbitMQ安裝

推薦使用docker安裝,不然要安裝Erlang語言環(huán)境相對麻煩
1、拉取鏡像
docker pull rabbitmq:management
2、啟動(dòng)RabbitMQ
docker run -d 以守護(hù)進(jìn)程方式在后臺(tái)運(yùn)行
--hostname rabbit_host1 設(shè)定容器的主機(jī)名(集群區(qū)分節(jié)點(diǎn))
--name xd_rabbit 指定容器名
-e RABBITMQ_DEFAULT_USER=admin 用戶名
-e RABBITMQ_DEFAULT_PASS=admin 密碼
-p 15672:15672 控制臺(tái)界面管理的端口映射
-p 5672:5672 程序訪問的端口映射
rabbitmq:management
3、查看啟動(dòng)日志
docker logs 返回的容器id
4、控制臺(tái)訪問(使用配置的用戶名和密碼)
http://www.lizhiyu.xyz:15672/

RabbitMQ使用

java依賴
簡單使用、不使用交換機(jī)、任務(wù)只會(huì)被一個(gè)消費(fèi)者消費(fèi)

RabbitMQ發(fā)送消息步驟圖

上圖說明如下 1、如上包括發(fā)送客戶端、RabbitMQ server、接收客戶端 2、RabbitMQ server中又包括 Virtual host(虛擬主機(jī))、exchange(交換機(jī))、queue(隊(duì)列) 3、虛擬主機(jī) 可以用于區(qū)分開發(fā)環(huán)境,發(fā)送端發(fā)送數(shù)據(jù)時(shí)要指定發(fā)送給哪個(gè)環(huán)境 4、隊(duì)列 用于存儲(chǔ)消息 5、交換機(jī) 交換機(jī)接收發(fā)送端發(fā)送的路由鍵(RoutingKey 最大長度255字節(jié)),根據(jù)接收的RoutingKey和配置的路由規(guī)則將發(fā)送的信息傳遞給指定隊(duì)列(起到路由的作用)使用說明如下 1、在RabbitMQ控制臺(tái)創(chuàng)建虛擬主機(jī)(一個(gè)虛擬主機(jī)下默認(rèn)有7個(gè)exchange) 2、創(chuàng)建交換機(jī)交換機(jī)類型有5種 direct，fanout，topic，headers，systemfanout 代表廣播類型(也就是綁定的所有隊(duì)列都會(huì)接收信息、這樣RoutingKey就不用了因不需要路由到指定隊(duì)列)direct 代表直連類型(可以通過指定的RouterKey路由到指定的隊(duì)列)例如:RouterKey為aaa則路由到aaa對應(yīng)的隊(duì)列topic 代表主題類型(可以通過一類的RouterKey進(jìn)行路由到指定的隊(duì)列)符號“#”匹配一個(gè)或多個(gè)詞，符號“*”匹配不多不少一個(gè)詞例如:RouterKey為aaa.*.* 或 aaa.* 會(huì)路由到aaa.#對應(yīng)的隊(duì)列RouterKey為aaa.* 會(huì)路由到aaa.*對應(yīng)的隊(duì)列topic相對于direct和fanout更加靈活不使用交換機(jī)，不設(shè)置交換機(jī)這樣就是用了默認(rèn)交換機(jī)(來達(dá)到消息只被消費(fèi)一次) 3、創(chuàng)建隊(duì)列設(shè)定隊(duì)列中存儲(chǔ)內(nèi)容持久化到本地(用于RabbitMQ崩潰后重啟隊(duì)列中已經(jīng)存儲(chǔ)的數(shù)據(jù)不丟失)設(shè)定隊(duì)列最大長度設(shè)定隊(duì)列是否獨(dú)占ttl說明:time to live 消息存活時(shí)間設(shè)定隊(duì)列中信息的ttl(隊(duì)列中信息過了指定時(shí)長沒有被消費(fèi)、就會(huì)處理這個(gè)消息)設(shè)置死信交換機(jī)(將隊(duì)列中的消息超過時(shí)長則將消息發(fā)送給死信交換機(jī)處理)當(dāng)消費(fèi)者拒收消息或隊(duì)列滿了或隊(duì)列中消息ttl達(dá)到極限則會(huì)將消息發(fā)送給信交換機(jī) 5、將交換機(jī)和隊(duì)列通過RoutingKey的規(guī)則進(jìn)行綁定(目的:設(shè)定路由規(guī)則)也就是說下次發(fā)送端發(fā)送過來的RoutingKey會(huì)根據(jù)設(shè)定的規(guī)則發(fā)送到指定隊(duì)列 6、發(fā)送端設(shè)定用戶名、密碼、端口、虛擬主機(jī)、以及RouteringKey、交換機(jī)、將消息發(fā)送

RabbitMQ廣播(Exchange的fanout類型使用)

RabbitMQ指定交換機(jī)給哪些隊(duì)列發(fā)送消息(Exchange的direct類型使用)

RabbitMQ根據(jù)消息類型將消息發(fā)送給指定隊(duì)列(Exchange的topic類型使用)

公平消費(fèi)和消息可靠性傳遞

消息可靠性傳遞要確保使用如下方法

一、接收端使用消息確認(rèn)機(jī)制(從而確認(rèn)消費(fèi)端是否消費(fèi)此任務(wù)) 二、RabbitMQ隊(duì)列消息持久化到內(nèi)存中(在RabbitMQ崩潰后重啟確保任務(wù)不丟失) 三、發(fā)送端發(fā)布確認(rèn)機(jī)制(在RabbitMQ崩潰時(shí)，有可能有些正在發(fā)布的任務(wù)還沒有持久化，還在內(nèi)存中從而導(dǎo)致消息丟失) 發(fā)布確認(rèn)分為兩個(gè)階段1.發(fā)送端成功將消息發(fā)送給交換機(jī)(exchange),exchange成功接收消息2.交換機(jī)(exchange)根據(jù)routerKey將消息發(fā)送給指定隊(duì)列，隊(duì)列成功接收消息3.如果消息發(fā)送失敗，失敗后會(huì)觸發(fā)監(jiān)聽然后重新發(fā)送消息(可以設(shè)置重新發(fā)送的次數(shù))

公平消費(fèi)的方法

RabbitMQ沒有辦法感知接收端正在處理的任務(wù)數(shù)量，僅僅是平均分配，這達(dá)不到我們想要的公平消費(fèi)，通過接收端沒有確認(rèn)的消息數(shù)量來判斷正在處理的任務(wù)數(shù)來達(dá)到公平消費(fèi)。

防止重復(fù)消費(fèi)

消息重復(fù)會(huì)重復(fù)的原因

1、在生產(chǎn)者將消息發(fā)送給交換機(jī)后，可能由于網(wǎng)絡(luò)原因生產(chǎn)者沒有收到交換機(jī)已經(jīng)收到消息的回復(fù)，導(dǎo)致生產(chǎn)者將同一條消息再次發(fā)送給交換機(jī) 2、隊(duì)列中的消息交給消費(fèi)者消費(fèi)過程中，可能由于網(wǎng)絡(luò)原因隊(duì)列沒有收到消費(fèi)者已經(jīng)消費(fèi)了此消息的回復(fù)，導(dǎo)致隊(duì)列中的消息再次被消費(fèi)者消費(fèi)

防止重復(fù)消費(fèi)

網(wǎng)絡(luò)問題無法解決需要消費(fèi)者處接口處理保持冪等性 1、根據(jù)消息的唯一id，使用redis 自增，如果增加到1則不進(jìn)行處理

有序消費(fèi)

有序消費(fèi) 場景1：一個(gè)生產(chǎn)者對應(yīng)一個(gè)消費(fèi)者

生產(chǎn)者為單節(jié)點(diǎn) 消息投放到同一個(gè)隊(duì)列 隊(duì)列對應(yīng)的消費(fèi)者只有一個(gè) 消費(fèi)者不開啟線程處理消息(如果開啟線程要求在消費(fèi)者處聲明多個(gè)內(nèi)存隊(duì)列,每個(gè)隊(duì)列中按順序存儲(chǔ)要消費(fèi)的消息)

有序消費(fèi) 場景2：一個(gè)生產(chǎn)者對應(yīng)多個(gè)消費(fèi)者

消費(fèi)者根據(jù)RouterKey將順序執(zhí)行的消息投放到同一個(gè)隊(duì)列中 每個(gè)隊(duì)列僅有一個(gè)消費(fèi)者進(jìn)行消費(fèi)

消息堆積怎么處理

首先看功能是否需要暫停，找到錯(cuò)誤原因 將舊的消息移動(dòng)到一個(gè)緩存隊(duì)列中，首先不影響新的消息被消費(fèi) 然后在慢慢消費(fèi)緩存隊(duì)列中的信息

spring集成RabbitMQ使用

springAMQP文檔
springAMQP詳細(xì)說明
自學(xué)代碼地址

RabbitMQ知識(shí)點(diǎn)

1、purge 清除(在清空隊(duì)列消息時(shí)候用到) 2、消息移動(dòng) 3、手動(dòng)確認(rèn)和自動(dòng)確認(rèn)自動(dòng)確認(rèn):可以配置重試次數(shù)，重試超過固定次數(shù)則拋棄或另外處理此消息手動(dòng)確認(rèn):隊(duì)列將消息發(fā)送給消費(fèi)者后，如果消費(fèi)者不回復(fù)隊(duì)列會(huì)一直任務(wù)此消息消費(fèi)者一直在執(zhí)行(必須要給隊(duì)列一個(gè)回復(fù)是否執(zhí)行成功,不然隊(duì)列會(huì)一直等著這條信息執(zhí)行完,除非消費(fèi)者掛掉了再分配此消息)手動(dòng)確認(rèn)如果沒有成功消費(fèi)時(shí)：可以用reids中的標(biāo)識(shí)來判斷是否要再次入隊(duì) 4、隊(duì)列排他獨(dú)占 5、通過死信隊(duì)列延時(shí)執(zhí)行任務(wù)時(shí)候,隊(duì)列和死信交換機(jī)綁定的ttl等參數(shù)消費(fèi)者重啟后不會(huì)覆蓋,需要重啟前刪除此隊(duì)列、或者沒有修改此隊(duì)隊(duì)列對應(yīng)參數(shù)

SpringBoot集成RabbitMQ源碼學(xué)習(xí)

參考

總結(jié)

以上是生活随笔為你收集整理的志宇-RabbitMQ学习的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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