RabbitMQ

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

RabbitMQ安裝

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

RabbitMQ使用

java依賴
簡單使用、不使用交換機、任務(wù)只會被一個消費者消費

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

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

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

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

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

公平消費和消息可靠性傳遞

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

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

公平消費的方法

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

防止重復(fù)消費

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

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

防止重復(fù)消費

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

有序消費

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

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

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

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

消息堆積怎么處理

首先看功能是否需要暫停，找到錯誤原因 將舊的消息移動到一個緩存隊列中，首先不影響新的消息被消費 然后在慢慢消費緩存隊列中的信息

spring集成RabbitMQ使用

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

RabbitMQ知識點

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

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

參考

總結(jié)

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

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