ActiveMQ、RabbitMQ、RocketMQ、Kafka四種訊息中間件分析介紹

我們從四種訊息中間件的介紹到基本使用，以及高可用，訊息重復性，訊息丟失，訊息順序性能方面進行分析介紹！

一、訊息中間件的使用場景

訊息中間件的使用場景總結就是六個字：解耦、異步、削峰

1.解耦

如果我方系統A要與三方B系統進行資料對接，推送系統人員資訊，通常我們會使用介面開發來進行，但是如果運維期間B系統進行了調整，或者推送程序中B系統網路進行了調整，又或者后續程序中我們需要推送資訊到三方C系統中，這樣的話就需要我們進行頻繁的介面開發調整，還需要考慮介面推送訊息失敗的場景，

如果我們使用訊息中間件進行訊息推送，我們只需要按照一種約定的資料結構進行資料推送，其他三方系統從訊息中間件取值消費就可以，即便是三方系統出現宕機或者其他調整，我們都可以正常進行資料推送，

總結：通過一個 MQ，Pub/Sub 發布訂閱訊息這么一個模型，A 系統就跟其它系統徹底解耦了，

2.異步

繼續我們上述的訊息推送業務，如果我們現在需要同時推送訊息到BCD三個系統中，而BCD系統接收到訊息后需要進行復雜的邏輯處理，以及讀庫寫庫處理，如果一個三方系統進行訊息處理需要1s，那我們遍歷推送完一次訊息，就需要三秒，

而如果我們使用訊息中間件，我們只需要推送到訊息中間件，然后進行介面回傳，可能只需要20ms，大大提升了用戶體驗，訊息推送后BCD系統各自進行訊息消費即可，不需要我們等待，

3.削峰

還是上述我們的應用場景，假設某一時間段內，每秒都有一條訊息推送，如果我們使用介面進行推送，BCD三個系統處理完就需要三秒，這樣會導致用戶前端體驗較差，而且系統后臺一直處于阻塞狀態，后續的訊息推送介面一直在等待，

如果我們使用訊息中間件，我們只需要將訊息推送至訊息中間件中，BCD系統對積壓的訊息進行相應的處理，

在上述高頻發的訊息時間段內，會在訊息中間中產生訊息積壓，BCD系統在上述時間段外對積壓訊息進行相應的處理即可，

二、訊息中間件的優缺點

訊息中間件的優點其實就是他的使用場景，

訊息中間件的缺點與優點也是相輔相成的，主要有以下三個

1.系統可用性降低

系統關聯的中間件越多，越容易引發宕機問題，

如上述案例中的問題，原本進行訊息推送我們只需要開發介面進行推送即可，引入訊息中間件后就需要考慮訊息中間件的高可用問題，如果訊息中間件出現宕機問題，我們所有的訊息推送都會失敗，

2.系統復雜度提高

上述案例中，如果我們使用介面進行訊息推送，我們只需要考慮介面超時以及介面推送訊息失敗的問題，但我們引入訊息中間件后，就需要考慮訊息中間件的維護，以及訊息重復消費，訊息丟失的問題，

3.一致性問題

上述案例中，如果我們使用介面進行訊息推送，推送訊息我們可以放在事務中處理，如果推送程序中出現例外，我們可以進行資料回滾，但我們引入訊息中間件后，就需要考慮訊息推送后，消費失敗的問題，以及如果我們同時推送訊息到BCD系統中，如何保證他們的事務一致性，

三、四種訊息中間件的基本介紹

1.ActiveMQ

1.1：Activemq 是什么

Activemq 是一種開源的，實作了JMS1.1規范的，面向訊息(MOM)的中間件，為應用程式提供高效的、可擴展的、穩定的和安全的企業級訊息通信，

1.2：Activemq 的作用及原理

Activemq 的作用就是系統之間進行通信，原理就是生產者生產訊息， 把訊息發送給activemq， Activemq 接收到訊息， 然后查看有多少個消費者，

然后把訊息轉發給消費者， 此程序中生產者無需參與， 消費者接收到訊息后做相應的處理和生產者沒有任何關系，

1.3：Activemq 的通信方式

publish(發布)-subscribe(訂閱)(發布-訂閱方式)

發布/訂閱方式用于多接收客戶端的方式，作為發布訂閱的方式，可能存在多個接收客戶端，并且接收端客戶端與發送客戶端存在時間上的依賴，一個接收端只能接收他創建以后發送客戶端發送的資訊，

p2p(point-to-point)(點對點)

p2p的程序則理解起來比較簡單，它好比是兩個人打電話，這兩個人是獨享這一條通信鏈路的，一方發送訊息，另外一方接收，就這么簡單，在實際應用中因為有多個用戶對使用p2p的鏈路，相互通信的雙方是通過一個類似于佇列的方式來進行交流，和前面pub-sub的區別在于一個topic有一個發送者和多個接收者，而在p2p里一個queue只有一個發送者和一個接收者，

1.4：Activemq 的訊息持久化機制

JDBC： 持久化到資料庫
AMQ ：日志檔案（已基本不用）
KahaDB ： AMQ基礎上改進，默認選擇
LevelDB ：谷歌K/V資料庫
在activemq.xml中查看默認的broker持久化機制，

1.5：Activemq 的訊息確認機制

（1）AUTO_ACKNOWLEDGE = 1 自動確認

（2）CLIENT_ACKNOWLEDGE = 2 客戶端手動確認

（3）DUPS_OK_ACKNOWLEDGE = 3 自動批量確認

（4）SESSION_TRANSACTED = 0 事務提交并確認

（5）INDIVIDUAL_ACKNOWLEDGE = 4 單條訊息確認

前四種是JMS API中提供的客戶端ACK_MODE，第五種是InforSuiteMQ自定義補充的一種ACK_MODE，

2.RabbitMQ

2.1：RabbitMQ是什么

RabbitMQ是一個由erlang語言撰寫的、開源的、在AMQP基礎上完整的、可復用的企業訊息系統，

2.2：RabbitMQ的作用及原理

基本概念

2.3：RabbitMQ的通信方式

2.3.1：簡單佇列

最簡單的作業佇列，其中一個訊息生產者，一個訊息消費者，一個佇列，也稱為點對點模式

2.3.2：作業佇列模式

一個訊息生產者，一個交換器，一個訊息佇列，多個消費者，同樣也稱為點對點模式

2.3.3：發布訂閱模式

Pulish/Subscribe，無選擇接收訊息，一個訊息生產者，一個交換機（交換機型別為fanout），多個訊息佇列，多個消費者

生產者只需把訊息發送到交換機，系結這個交換機的佇列都會獲得一份一樣的資料，

2.3.4：路由模式

在發布/訂閱模式的基礎上，有選擇的接收訊息，也就是通過 routing 路由進行匹配條件是否滿足接收訊息，

2.3.5：主體模式

topics(主題)模式跟routing路由模式類似，只不過路由模式是指定固定的路由鍵 routingKey，而主題模式是可以模糊匹配路由鍵 routingKey，類似于SQL中 = 和 like 的關系，

2.3.6：RPC模式

與上面其他5種所不同之處，該模式是擁有請求/回復的，也就是有回應的，上面5種都沒有，

RPC是指遠程程序呼叫，也就是說兩臺服務器A，B，一個應用部署在A服務器上，想要呼叫B服務器上應用提供的處理業務，處理完后然后在A服務器繼續執行下去，把異步的訊息以同步的方式執行，

2.4：RabbitMQ的訊息持久化機制

Queue（訊息佇列）的持久化是通過durable=true來實作的，

Message（訊息）的持久化 ，通過設定訊息是持久化的標識，

Exchange（交換機）的持久化 ，

2.5：RabbitMQ的訊息確認機制

confirm機制：確認訊息是否成功發送到Exchange

ack機制：確認訊息是否被消費者成功消費

• AcknowledgeMode.NONE：自動確認
• AcknowledgeMode.AUTO：根據情況確認
• AcknowledgeMode.MANUAL：手動確認

3.RocketMQ

3.1：RocketMQ是什么

RocketMQ是阿里開發的一款純java、分布式、佇列模型的開源訊息中間件，支持事務訊息、順序訊息、批量訊息、定時訊息、訊息回溯等，

3.2：RocketMQ的作用及原理

基本概念

3.3：RocketMQ的通信方式

RocketMQ訊息訂閱有兩種模式

一種是Push模式（MQPushConsumer），即MQServer主動向消費端推送

另外一種是Pull模式（MQPullConsumer），即消費端在需要時，主動到MQ Server拉取

但在具體實作時，Push和Pull模式本質都是采用消費端主動拉取的方式，即consumer輪詢從broker拉取訊息

集群模式和廣播模式

集群模式：默認情況下我們都是使用的集群模式，也就是說消費者組收到訊息后，只有其中的一臺機器會接收到訊息，

廣播模式：消費者組內的每臺機器都會收到這條訊息，

3.4：RocketMQ的訊息持久化機制

exchange持久化、queue持久化、message持久化

CommitLog：日志資料檔案，存盤訊息內容，所有 queue 共享，不區分 topic ，順序讀寫 ，1G 一個檔案

ConsumeQueue：邏輯 Queue，基于 topic 的 CommitLog 的索引檔案，訊息先到達 commitLog，然后異步轉發到 consumeQueue，包含 queue 在 commitLog 中的物理偏移量 offset，訊息物體內容大小和 Message Tag 的 hash 值，大于 600W 個位元組，寫滿之后重新生成，順序寫

IndexFile：基于 Key 或 時間區間的 CommitLog 的索引檔案，檔案名以創建的時間戳命名，固定的單個 indexFile 大小為 400M，可以保存 2000W 個索引

3.5：RocketMQ的訊息確認機制

confirm機制：確認訊息是否成功發送到Exchange

ack機制：確認訊息是否被消費者成功消費

4.Kafka

4.1：Kafka是什么

Kafka是由Apache軟體基金會開發的一個開源流處理平臺，由Scala和Java撰寫，Kafka是一個分布式、磁區的、多副本的、多訂閱者，基于zookeeper協調的分布式日志系統，可作為訊息中間件

4.2：Kafka的作用及原理

基本概念

4.3：Kafka的通信方式

生產者發送模式

1.發后即忘（fire-and-forget）：只管往Kafka中發送訊息而并不關心訊息是否正確到達

2.同步（sync）：一般是在send（）方法里指定一個Callback的回呼函式，Kafka在回傳回應時呼叫該函式來實作異步的發送確認，

3.異步（async）：send()方法會回傳Futrue物件，通過呼叫Futrue物件的get()方法，等待直到結果回傳

消費者消費模式

1.At-most-once（最多一次）：在每一條訊息commit成功之后，再進行消費處理;設定自動提交為false，接收到訊息之后，首先commit，然后再進行消費，

2.At-least-once（最少一次）：在每一條訊息處理成功之后，再進行commit;設定自動提交為false;訊息處理成功之后，手動進行commit，

3.Exactly-once（正好一次）：將offset作為唯一id與訊息同時處理，并且保證處理的原子性;設定自動提交為false;訊息處理成功之后再提交，

4.4：Kafka的訊息持久化機制

Kafka直接將資料寫入到日志檔案中，以追加的形式寫入

4.5：Kafka的訊息確認機制

confirm機制：確認訊息是否成功發送

ack機制：確認訊息是否被消費者成功消費

四、訊息佇列高可用

引言：系統應用MQ作為訊息中間件后，會導致系統可用性降低，所以只要你用了 MQ，高可用肯定是要考慮到的

1.ActiveMQ高可用

ActiveMQ的部署方式有三種，分別為：單節點部署（不支持高可用），Master-Slave部署方式（主從模式），Broker-Cluster部署方式（負載均衡）

1.1.單節點部署（不支持高可用）

單節點部署方式因為不支持高可用，只會在開發或者測驗環境下用到，且單節點部署方式較簡單，不進行詳述，

1.2.Master-Slave部署方式（支持高可用）

1.2.1.shared filesystem Master-Slave部署方式

主要是通過共享存盤目錄來實作master和slave的熱備，所有的ActiveMQ應用都在不斷地獲取共享目錄的控制權，哪個應用搶到了控制權，它就成為master，

多個共享存盤目錄的應用，誰先啟動，誰就可以最早取得共享目錄的控制權成為master，其他的應用就只能作為slave，

1.2.2.shared database Master-Slave方式

與shared filesystem方式類似，只是共享的存盤介質由檔案系統改成了資料庫而已，

1.2.3.Replicated LevelDB Store方式

這種主備方式是ActiveMQ5.9以后才新增的特性，使用ZooKeeper協調選擇一個node作為master，被選擇的master broker node開啟并接受客戶端連接，

其他node轉入slave模式，連接master并同步他們的存盤狀態，slave不接受客戶端連接，所有的存盤操作都將被復制到連接至Master的slaves，

如果master死了，得到了最新更新的slave被允許成為master，fialed node能夠重新加入到網路中并連接master進入slave mode，所有需要同步的disk的訊息操作都將等待存盤狀態被復制到其他法定節點的操作完成才能完成，

當一個新的master被選中，你需要至少保障一個法定node在線以能夠找到擁有最新狀態的node，這個node將會成為新的master，因此，推薦運行至少3個replica nodes，以防止一個node失敗了，服務中斷，

1.3.Broker-Cluster部署方式（不支持高可用）

前面的Master-Slave的方式雖然能解決多服務熱備的高可用問題，但無法解決負載均衡和分布式的問題，Broker-Cluster的部署方式就可以解決負載均衡的問題，

? Broker-Cluster部署方式中，各個broker通過網路互相連接，并共享queue，當broker-A上面指定的queue-A中接收到一個message處于pending狀態，而此時沒有consumer連接broker-A時，如果cluster中的broker-B上面由一個consumer在消費queue-A的訊息，那么broker-B會先通過內部網路獲取到broker-A上面的message，并通知自己的consumer來消費，

1.3.1.static Broker-Cluster部署

在activemq.xml檔案中靜態指定Broker需要建立橋連接的其他Broker

1.3.2.Dynamic Broker-Cluster部署

在activemq.xml檔案中不直接指定Broker需要建立橋連接的其他Broker，由activemq在啟動后動態查找

1.4.Master-Slave與Broker-Cluster相結合的部署方式

可以看到Master-Slave的部署方式雖然解決了高可用的問題，但不支持負載均衡，

Broker-Cluster解決了負載均衡，但當其中一個Broker突然宕掉的話，那么存在于該Broker上處于Pending狀態的message將會丟失，無法達到高可用的目的，

Master-Slave與Broker-Cluster相結合的部署方式是目前ActiveMQ比較推薦的部署方案，

2.RabbitMQ高可用

RabbitMQ的部署方式有三種，分別為：單機模式（不支持高可用），普通集群模式（不支持高可用），鏡像集群模式（支持高可用）

2.1單機模式（不支持高可用）

單節點部署方式因為不支持高可用，只會在開發或者測驗環境下用到，且單節點部署方式較簡單，不進行詳述，

2.2普通集群模式（不支持高可用）

普通集群模式，意思就是在多臺機器上啟動多個 RabbitMQ 實體，每個機器啟動一個，你創建的 queue，只會放在一個 RabbitMQ 實體上，但是每個實體都同步 queue 的元資料（元資料可以認為是 queue 的一些配置資訊，通過元資料，可以找到 queue 所在實體），你消費的時候，實際上如果連接到了另外一個實體，那么那個實體會從 queue 所在實體上拉取資料過來，

這種方式確實很麻煩，也不怎么好，沒做到所謂的分布式，就是個普通集群，因為這導致你要么消費者每次隨機連接一個實體然后拉取資料，要么固定連接那個 queue 所在實體消費資料，前者有資料拉取的開銷，后者導致單實體性能瓶頸，

而且如果那個放 queue 的實體宕機了，會導致接下來其他實體就無法從那個實體拉取，如果你開啟了訊息持久化，讓 RabbitMQ 落地存盤訊息的話，訊息不一定會丟，得等這個實體恢復了，然后才可以繼續從這個 queue 拉取資料，

2.3鏡像集群模式（支持高可用）

這種模式，才是所謂的 RabbitMQ 的高可用模式，跟普通集群模式不一樣的是，在鏡像集群模式下，你創建的 queue，無論元資料還是 queue 里的訊息都會存在于多個實體上，就是說，每個 RabbitMQ 節點都有這個 queue 的一個完整鏡像，包含 queue 的全部資料的意思，然后每次你寫訊息到 queue 的時候，都會自動把訊息同步到多個實體的 queue 上，

3.RocketMQ高可用

RocketMQ的部署方式有兩種，分別為：單節點模式（不支持高可用），多節點模式

3.1.單節點模式（不支持高可用）

單節點部署方式因為不支持高可用，只會在開發或者測驗環境下用到，且單節點部署方式較簡單，不進行詳述，

3.2.多節點模式

3.2.1.多Master模式（不支持高可用）

一個集群無 Slave，全是 Master，例如 2 個 Master 或者 3 個 Master

配置簡單，單個Master 宕機或重啟維護對應用無影響，

單臺機器宕機期間，這臺機器上未被消費的訊息在機器恢復之前不可訂閱，訊息實時性會受到受到影響，

3.2.2.多Master多Slave模式（異步復制）（支持高可用）

每個 Master 配置一個 Slave，有多對Master-Slave， HA，采用異步復制方式，主備有短暫訊息延遲，毫秒級，

即使磁盤損壞，訊息丟失的非常少，且訊息實時性不會受影響，因為Master 宕機后，消費者仍然可以從 Slave消費，此程序對應用透明，不需要人工干預，性能同多 Master 模式幾乎一樣，

Master 宕機，磁盤損壞情況，會丟失少量訊息，

3.2.3.多Master多Slave模式（同步雙寫）（支持高可用）

每個 Master 配置一個 Slave，有多對Master-Slave， HA采用同步雙寫方式，主備都寫成功，向應用回傳成功，

資料與服務都無單點， Master宕機情況下，訊息無延遲，服務可用性與資料可用性都非常高

性能比異步復制模式略低，大約低 10%左右，發送單個訊息的 RT會略高，

4.Kafka高可用

Kafka的部署方式有三種，分別為：單broke節點（不支持高可用），單機多broker模式（支持高可用），多機多broker模式（支持高可用）

4.1.單broke節點（不支持高可用）

單節點部署方式因為不支持高可用，只會在開發或者測驗環境下用到，且單節點部署方式較簡單，不進行詳述，

4.2.單機多broker模式（支持高可用）

這種部署方式其實是一種偽集群模式，單機部署多節點如果出現服務器宕機，那么所有節點都不能正常提供服務，

4.3.多機多broker模式（支持高可用）

Kafka 0.8 以前，是沒有 HA 機制的，就是任何一個 broker 宕機了，那個 broker 上的 partition 就廢了，沒法寫也沒法讀，沒有什么高可用性可言，

Kafka 0.8 以后，提供了 HA 機制，就是 replica（復制品） 副本機制，每個 partition 的資料都會同步到其它機器上，形成自己的多個 replica 副本，所有 replica 會選舉一個 leader 出來，那么生產和消費都跟這個 leader 打交道，然后其他 replica 就是 follower，寫的時候，leader 會負責把資料同步到所有 follower 上去，讀的時候就直接讀 leader 上的資料即可，只能讀寫 leader？很簡單，要是你可以隨意讀寫每個 follower，那么就要 care 資料一致性的問題，系統復雜度太高，很容易出問題，Kafka 會均勻地將一個 partition 的所有 replica 分布在不同的機器上，這樣才可以提高容錯性，

五、訊息重復消費問題

引言：為什么要考慮重復消費的問題？比如我們消費后通過消費中間件來呼叫，扣費10元，但是消費者消費訊息后還沒來得及進行確認，訊息中間件進行了重啟，那么訊息者就會進行再次扣費處理，這樣就會出問題！

ActiveMQ、RabbitMQ、RocketMQ、Kafka，都有可能會出現訊息重復消費的問題，正常，因為這問題通常不是 MQ 自己保證的，是由我們開發來保證的，

我們以Kafka為例說明一下重復消費的問題：

Kafka 實際上有個 offset 的概念，就是每個訊息寫進去，都有一個 offset，代表訊息的序號，然后 consumer 消費了資料之后，每隔一段時間（定時定期），會把自己消費過的訊息的 offset 提交一下，表示“我已經消費過了，下次我要是重啟啥的，你就讓我繼續從上次消費到的 offset 來繼續消費吧”，

但是，如果在這期間重啟系統或者直接 kill 行程了，再重啟，這會導致 consumer 有些訊息處理了，但是沒來得及提交 offset，重啟之后，少數訊息會再次消費一次，

如果消費者干的事兒是拿一條資料就往資料庫里寫一條，會導致說，你可能就把資料在資料庫里插入了 2 次，那么資料就錯啦，

重復消費問題引發后，我們就需要考慮怎么保證冪等性，

冪等性，通俗點說，就一個資料，或者一個請求，給你重復來多次，你得確保對應的資料是不會改變的，不能出錯，

保證冪等性的具體實作方式需要結合對應的業務去實作，這里提供幾個思路：

• 如果是資料插入操作，插入前我們根據唯一鍵先進行查詢，如果已有資料那我們只進行更新就行，
• 如果是寫 Redis，則我們無需考慮冪等性，反正每次都是 set，天然冪等性，
• 如果是基于資料庫的唯一鍵來保證重復資料不會重復插入多條，因為有唯一鍵約束了，重復資料插入只會報錯，不會導致資料庫中出現臟資料，
• 如果不是以上集中通用的場景，那需要我們發送訊息的時候攜帶唯一ID，消費者在消費前進行相應的查重處理，處理后在進行相應的業務操作，

六、訊息丟失問題

引言：MQ 有個基本原則，就是資料不能多一條，也不能少一條，

不能多，就是上面說的重復消費和冪等性問題，

不能少，就是說這資料別搞丟了，那這個問題你必須得考慮一下，

訊息丟失的問題需要從生產者、MQ、消費者三個方面來進行考慮，相應的解決方案也需要從這三方面出發（生產者確認機制，MQ訊息持久化、消費者確認機制），

1.ActiveMQ

1.生產者丟失訊息

生產者丟失訊息的問題可以通過訊息重投、重試機制來解決

2.ActiveMQ丟失訊息

ActiveMQ丟失訊息的問題需要通過ActiveMQ訊息持久化機制+高可用（見ActiveMQ章節）來解決，ActiveMQ的訊息持久化機制有以下幾種

JDBC： 持久化到資料庫
AMQ ：日志檔案（已基本不用）
KahaDB ： AMQ基礎上改進，默認選擇
LevelDB ：谷歌K/V資料庫
在activemq.xml中查看默認的broker持久化機制，

3.訊息者丟失訊息

消費者丟失訊息通過ack機制來解決，訊息者進行業務處理后，再進行ack確認，避免訊息丟失，

2.RabbitMQ

1.生產者丟失訊息

生產者訊息丟失，通過confirm機制來確認訊息發送，然后進行相應的訊息重投、重試機制

2.RabbitMQ丟失訊息

RabbitMQ丟失訊息的問題需要通過RabbitMQ訊息持久化機制+高可用（見RabbitMQ章節）來解決，

RabbitMQ持久化包含：

Queue（訊息佇列）的持久化是通過durable=true來實作的，

Message（訊息）的持久化 ，通過設定訊息是持久化的標識，

Exchange（交換機）的持久化 ，

3.訊息者丟失訊息

消費者丟失訊息通過ack機制來解決，訊息者進行業務處理后，再進行ack確認，避免訊息丟失，

3.RocketMQ

1.生產者丟失訊息

生產者訊息丟失，通過confirm機制來確認訊息發送，然后進行相應的訊息重投、重試機制

2.RocketMQ丟失訊息

RocketMQ丟失訊息的問題需要通過RocketMQ訊息持久化機制+高可用（見RocketMQ章節）來解決，

RocketMQ持久化包含：exchange持久化、queue持久化、message持久化

3.訊息者丟失訊息

消費者丟失訊息通過ack機制來解決，訊息者進行業務處理后，再進行ack確認，避免訊息丟失，

4.Kafka

1.生產者丟失訊息

生產者訊息丟失，通過confirm機制來確認訊息發送，然后進行相應的訊息重投、重試機制

2.Kafka丟失訊息

Kafka直接將資料寫入到日志檔案中，以追加的形式寫入

3.訊息者丟失訊息

消費者丟失訊息通過ack機制來解決，訊息者進行業務處理后，再進行ack確認，避免訊息丟失，

總結：其實MQ訊息丟失，無非就是生產者發送時丟失，MQ傳遞時丟失，消費者消費時丟失幾種問題，我們相應的從以上三方面解決就可以，但是上述三種方式使用后，其實也不能保證100%訊息不丟失，所以往往在業務場景還會使用資料庫輔助記錄的方式，來保證訊息不丟失，但資料庫輔助記錄方式對相關性能以及使用用較大的影響，所以一般資料只需要進行上面三種方式處理，就能保證訊息基本不丟失，發生訊息丟失時我們配合日志進行相應的訊息恢復就可以，

資料庫輔助記錄：生產者發送訊息時同步發送一條訊息到資料庫中，消費者拿到訊息并完成業務處理后，從資料庫洗掉對應的記錄，

七、訊息順序性問題

引言：為什么要保證訊息的順序性？

比如現在我們有個賬號余額為5，我們充值50元，購買一件20元的商品，但因訊息不能保證順序，導致先進行扣費處理，這樣就會導致我們購買失敗，

訊息順序性消費情況，尤其在高可用（集群方式）下一定要考慮，

1.ActiveMQ

ActiveMQ因為默認是單queue 佇列，所以它模式就是保證訊息順序性消費的，

2.RabbitMQ

• 將RabbitMQ拆分多個 queue，每個 queue 一個 consumer，保證訊息的順序性，
• 一個 queue 但是對應一個 consumer，然后這個 consumer 內部用記憶體佇列做排隊，然后分發給底層不同的 worker 來處理，

3.RocketMQ

RocketMQ保證訊息順序性方法與Kafka大致相同，

• 一個 topic，一個 queue，一個 consumer，內部單執行緒消費，單執行緒吞吐量太低，一般不會用這個，
• 寫 N 個記憶體 queue，具有相同 key 的資料都到同一個記憶體 queue；然后對于 N 個執行緒，每個執行緒分別消費一個記憶體 queue 即可，這樣就能保證順序性，

4.Kafka

• 一個 topic，一個 partition，一個 consumer，內部單執行緒消費，單執行緒吞吐量太低，一般不會用這個，
• 寫 N 個記憶體 queue，具有相同 key 的資料都到同一個記憶體 queue；然后對于 N 個執行緒，每個執行緒分別消費一個記憶體 queue 即可，這樣就能保證順序性，

八、訊息積壓問題

引言：如何解決訊息佇列的延時以及過期失效問題？訊息佇列滿了以后該怎么處理？有幾百萬訊息持續積壓幾小時，怎么處理？

其實訊息積壓的問題，一般都是由消費端出了問題導致的，在實際業務場景中一般不會出現，但是出現問題一般都是大問題，

模擬場景：

一個消費者一秒是 1000 條，一秒 3 個消費者是 3000 條，一分鐘就是 18 萬條，由于消費者宕機導致現在MQ中積壓幾百萬資料

解決思路：

• 先修復 consumer 的問題，確保其恢復消費速度，然后將現有 consumer 都停掉（避免重復消費），
• 新建一個 topic，partition 是原來的 10 倍，臨時建立好原先 10 倍的 queue 數量，
• 然后寫一個臨時的分發資料的 consumer 程式，這個程式部署上去消費積壓的資料，消費之后不做耗時的處理，直接均勻輪詢寫入臨時建立好的 10 倍數量的 queue，
• 接著臨時征用 10 倍的機器來部署 consumer，每一批 consumer 消費一個臨時 queue 的資料，這種做法相當于是臨時將 queue 資源和 consumer 資源擴大 10 倍，以正常的 10 倍速度來消費資料，
• 等快速消費完積壓資料之后，得恢復原先部署的架構，重新用原先的 consumer 機器來消費訊息，

mq 中的訊息過期失效了

假設你用的是 RabbitMQ，RabbtiMQ 是可以設定過期時間的，也就是 TTL，如果訊息在 queue 中積壓超過一定的時間就會被 RabbitMQ 給清理掉，這個資料就沒了，

假設 1 萬個訂單積壓在 mq 里面，沒有處理，其中 1000 個訂單都丟了，你只能手動寫程式把那 1000 個訂單給查出來，手動發到 mq 里去再補一次，

mq 都快寫滿了

如果訊息積壓在 mq 里，長時間都沒有處理掉，此時導致 mq 都快寫滿了，咋辦？

這種情況下只能是通過增加臨時Consumer將資料進行快速消費，等MQ恢復正常后再補充資料，

RocketMQ方案

對于 RocketMQ，官方針對訊息積壓問題，提供了解決方案，

1. 提高消費并行度
   絕大部分訊息消費行為都屬于 IO 密集型，即可能是操作資料庫，或者呼叫 RPC，這類消費行為的消費速度在于后端資料庫或者外系統的吞吐量，通過增加消費并行度，可以提高總的消費吞吐量，但是并行度增加到一定程度，反而會下降，所以，應用必須要設定合理的并行度， 如下有幾種修改消費并行度的方法：

? 同一個 ConsumerGroup 下，通過增加 Consumer 實體數量來提高并行度（需要注意的是超過訂閱佇列數的 Consumer 實體無效），可以通過加機器，或者 在已有機器啟動多個行程的方式， 提高單個 Consumer 的消費并行執行緒，通過修改引數 consumeThreadMin、consumeThreadMax 實作，

2. 批量方式消費
   某些業務流程如果支持批量方式消費，則可以很大程度上提高消費吞吐量，例如訂單扣款類應用，一次處理一個訂單耗時 1 s，一次處理 10 個訂單可能也只耗時 2 s，這樣即可大幅度提高消費的吞吐量，通過設定 consumer 的 consumeMessageBatchMaxSize 返個引數，默認是 1，即一次只消費一條訊息，例如設定為 N，那么每次消費的訊息數小于等于 N，

3. 跳過非重要訊息
   發生訊息堆積時，如果消費速度一直追不上發送速度，如果業務對資料要求不高的話，可以選擇丟棄不重要的訊息，例如，當某個佇列的訊息數堆積到 100000 條以上，則嘗試丟棄部分或全部訊息，這樣就可以快速追上發送訊息的速度，示例代碼如下：

4. 優化每條訊息消費程序
   舉例如下，某條訊息的消費程序如下：

九、自我實作訊息佇列思路

引言：如果讓你寫一個訊息佇列，該如何進行架構設計？

比如說訊息佇列系統，我們從以下幾個角度來考慮一下：

• 可擴展性：就是需要的時候快速擴容，就可以增加吞吐量和容量，可以參考afka 的設計理念，broker -> topic -> partition，每個 partition 放一個機器，就存一部分資料，
• 持久化：為了保證MQ的訊息不丟失，設計時一定要考慮訊息的持久化機制，且持久化要順序寫，這樣就沒有磁盤隨機讀寫的尋址開銷，磁盤順序讀寫的性能是很高的，這就是 kafka 的思路，
• 高可用：保證MQ的可靠性，可以參考kafka ，多副本 -> leader & follower -> broker 掛了重新選舉 leader 即可對外服務，
• 能不能支持資料 0 丟失啊？可以的，參考我們之前說的那個 kafka 資料零丟失方案，

十、MQ總結

其實MQ的使用，無非就是從原理，高可用，重復訊息，順序讀寫，資料丟失幾個方面開展，

上述的介紹是偏重思路方面來進行展開的，至于具體的MQ使用細節，我想你有了對應的思路去查會有一大堆，這也是我學習技術的一個思路，先掌握一個大的方向，然后沿著一個大的方向再進行相應的詳細學習，

最后，上述MQ介紹中，大部分都是有我平時開發積累所得，也有一部分是借助網路現場學習，

如有不足或錯誤，歡迎大家指出，我們共同學習進步！

標籤：其他

上一篇：HttpClient如何進行RequestConfig的相關配置呢？

下一篇：ngnix+tomcat轉發、負載均衡