Kafka、RabbitMQ、RocketMQ 之間的區(qū)別是什么？

性能

消息中間件的性能主要衡量吞吐量，Kafka的吞吐量比RabbitMQ要高出1~2個(gè)數(shù)量級(jí)，RabbitMQ的單機(jī)QPS在萬(wàn)級(jí)別，Kafka的單機(jī)QPS能夠達(dá)到百萬(wàn)級(jí)別。RocketMQ單機(jī)寫(xiě)入TPS單實(shí)例約7萬(wàn)條/秒，單機(jī)部署3個(gè)Broker，可以跑到最高12萬(wàn)條/秒，消息大小10個(gè)字節(jié)，Kafka如果開(kāi)啟冪等、事務(wù)等功能，性能也會(huì)有所降低。

數(shù)據(jù)可靠性

Kafka與RabbitMQ 都具備多副本機(jī)制，數(shù)據(jù)可靠性較高。RocketMQ支持異步實(shí)時(shí)刷盤(pán)，同步刷盤(pán)，同步Replication（拷貝），異步Replication，數(shù)據(jù)可靠性高。

服務(wù)可用性

Kafka采用集群部署，分區(qū)與多副本的設(shè)計(jì)，使得單節(jié)點(diǎn)宕機(jī)對(duì)服務(wù)無(wú)影響，并且支持消息容量的線(xiàn)性提升。RabbitMQ支持集群部署，集群節(jié)點(diǎn)數(shù)量有多種規(guī)格。RocketMQ是分布式架構(gòu)，可用性高。

功能

Kafka與RabbitMQ都是比較主流的兩款消息中間件，具備消息傳遞的基本功能，但在一些特殊的功能方面存在差異，RocketMQ在阿里集團(tuán)內(nèi)部有大量的應(yīng)用在使用。

RabbitMQ如何保證消息不丟失？

Kafka 的架構(gòu)說(shuō)一下？

整個(gè)架構(gòu)中包括三個(gè)角色：
? 生產(chǎn)者（Producer）：消息和數(shù)據(jù)生產(chǎn)者。
? 代理（Broker）：緩存代理，Kafka的核心功能。
? 消費(fèi)者（Consumer）：消息和數(shù)據(jù)消費(fèi)者。
Kafka給Producer和Consumer提供注冊(cè)的接口，數(shù)據(jù)從Producer發(fā)送到Broker，Broker承擔(dān)一個(gè)中間緩存和分發(fā)的作用，負(fù)責(zé)分發(fā)注冊(cè)到系統(tǒng)中的Consumer。

Kafka 怎么保證消息是有序的？

消息在被追加到 Partition(分區(qū))的時(shí)候都會(huì)分配一個(gè)特定的偏移量（offset）。Kafka 通過(guò)偏移量（offset）來(lái)保證消息在分區(qū)內(nèi)的有序性。發(fā)送消息的時(shí)候指定key/Partition。

Kafka 怎么解決重復(fù)消費(fèi)？

? 生產(chǎn)者發(fā)送每條數(shù)據(jù)的時(shí)候，里面加一個(gè)全局唯一的id，消費(fèi)到了之后，先根據(jù)這個(gè)id去比對(duì)Redis里查一下，之前是否消費(fèi)過(guò)，如果沒(méi)有消費(fèi)過(guò)，就處理，然后把這個(gè)id寫(xiě)入到Redis中。如果消費(fèi)過(guò)就別處理了。
? 基于數(shù)據(jù)庫(kù)的唯一鍵來(lái)保證重復(fù)數(shù)據(jù)不會(huì)重復(fù)插入多條。因?yàn)橛形ㄒ绘I約束了，重復(fù)數(shù)據(jù)插入只會(huì)報(bào)錯(cuò)，不會(huì)導(dǎo)致數(shù)據(jù)庫(kù)中出現(xiàn)臟數(shù)據(jù)。

Kafka 怎么保證消息不丟失？

生產(chǎn)者丟失消息的情況

生產(chǎn)者(Producer) 調(diào)用send方法發(fā)送消息之后，消息可能因?yàn)榫W(wǎng)絡(luò)問(wèn)題并沒(méi)有發(fā)送過(guò)去。 為了確定消息是發(fā)送成功，我們要判斷消息發(fā)送的結(jié)果，Kafka 生產(chǎn)者(Producer) 使用 send 方法發(fā)送消息實(shí)際上是異步的操作，我們可以通過(guò) get()方法獲取調(diào)用結(jié)果，但是這樣也讓它變?yōu)榱送讲僮?#xff0c;可以采用為其添加回調(diào)函數(shù)的形式，示例代碼如下：
ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, o);
future.addCallback(result -> logger.info(“生產(chǎn)者成功發(fā)送消息到topic:{} partition:{}的消息”, result.getRecordMetadata().topic(), result.getRecordMetadata().partition()),
ex -> logger.error(“生產(chǎn)者發(fā)送消失敗，原因：{}”, ex.getMessage())); Producer 的retries（重試次數(shù)）設(shè)置一個(gè)比較合理的值，一般是 3 ，但是為了保證消息不丟失的話(huà)一般會(huì)設(shè)置比較大一點(diǎn)。設(shè)置完成之后，當(dāng)出現(xiàn)網(wǎng)絡(luò)問(wèn)題之后能夠自動(dòng)重試消息發(fā)送，避免消息丟失。另外，建議還要設(shè)置重試間隔，因?yàn)殚g隔太小的話(huà)重試的效果就不明顯了，網(wǎng)絡(luò)波動(dòng)一次你3次一下子就重試完了

消費(fèi)者丟失消息的情況
當(dāng)消費(fèi)者拉取到了分區(qū)的某個(gè)消息之后，消費(fèi)者會(huì)自動(dòng)提交了 offset。自動(dòng)提交的話(huà)會(huì)有一個(gè)問(wèn)題，試想一下，當(dāng)消費(fèi)者剛拿到這個(gè)消息準(zhǔn)備進(jìn)行真正消費(fèi)的時(shí)候，突然掛掉了，消息實(shí)際上并沒(méi)有被消費(fèi)，但是 offset 卻被自動(dòng)提交了。
解決辦法也比較粗暴，我們手動(dòng)關(guān)閉自動(dòng)提交 offset，每次在真正消費(fèi)完消息之后再自己手動(dòng)提交 offset 。 但是，細(xì)心的朋友一定會(huì)發(fā)現(xiàn)，這樣會(huì)帶來(lái)消息被重新消費(fèi)的問(wèn)題。比如你剛剛消費(fèi)完消息之后，還沒(méi)提交 offset，結(jié)果自己掛掉了，那么這個(gè)消息理論上就會(huì)被消費(fèi)兩次。

Kafka 弄丟了消息
試想一種情況：假如 leader 副本所在的 broker 突然掛掉，那么就要從 follower 副本重新選出一個(gè) leader ，但是 leader 的數(shù)據(jù)還有一些沒(méi)有被 follower 副本的同步的話(huà)，就會(huì)造成消息丟失。
當(dāng)我們配置了 unclean.leader.election.enable = false 的話(huà)，當(dāng) leader 副本發(fā)生故障時(shí)就不會(huì)從 follower 副本中和 leader 同步程度達(dá)不到要求的副本中選擇出 leader ，這樣降低了消息丟失的可能性。

RocketMQ 如何監(jiān)聽(tīng)消息的？

RocketMQ 常見(jiàn)的面試題

鏈接1： (20條消息) RocketMQ 常見(jiàn)面試問(wèn)題_rocketmq面試題_柚子茶1990的博客-CSDN博客 主要是保證消息的不丟失。

哪個(gè)環(huán)節(jié)會(huì)有消息丟失的可能？

我們考慮一個(gè)通用的MQ場(chǎng)景：

其中，1，2，4三個(gè)場(chǎng)景都是跨網(wǎng)絡(luò)的，而跨網(wǎng)絡(luò)就肯定會(huì)有丟消息的可能。 然后關(guān)于3這個(gè)環(huán)節(jié)，通常MQ存盤(pán)時(shí)都會(huì)先寫(xiě)入操作系統(tǒng)的緩存page cache中，然后再由操作系統(tǒng)異步的將消息寫(xiě)入硬盤(pán)。這個(gè)中間有個(gè)時(shí)間差，就可能會(huì)造成消息丟失。如果服務(wù)掛了，緩存中還沒(méi)有來(lái)得及寫(xiě)入硬盤(pán)的消息就會(huì)丟失。 這個(gè)是MQ場(chǎng)景都會(huì)面對(duì)的通用的丟消息問(wèn)題。那我們看看用Rocket時(shí)要如何解決這個(gè)問(wèn)題

RocketMQ消息零丟失方案

1、生產(chǎn)者使用事務(wù)消息機(jī)制保證消息零丟失

這個(gè)結(jié)論比較容易理解，因?yàn)镽ocketMQ的事務(wù)消息機(jī)制就是為了保證零丟失來(lái)設(shè)計(jì)的，并且經(jīng)過(guò)阿里的驗(yàn)證，肯定是非?？孔V的。 但是如果深入一點(diǎn)的話(huà)，我們還是要理解下這個(gè)事務(wù)消息到底是不是靠譜。我們以最常見(jiàn)的電商訂單場(chǎng)景為例，來(lái)簡(jiǎn)單分析下事務(wù)消息機(jī)制如何保證消息不丟失。我們看下下面這個(gè)流程圖：

（1）、為什么要發(fā)送個(gè)half消息？有什么用？

這個(gè)half消息是在訂單系統(tǒng)進(jìn)行下單操作前發(fā)送，并且對(duì)下游服務(wù)的消費(fèi)者是不可見(jiàn)的。那這個(gè)消息的作用更多的體現(xiàn)在確認(rèn)RocketMQ的服務(wù)是否正常。相當(dāng)于嗅探下RocketMQ服務(wù)是否正常，并且通知RocketMQ，我馬上就要發(fā)一個(gè)很重要的消息了，你做好準(zhǔn)備。

（2）.half消息如果寫(xiě)入失敗了怎么辦？

如果沒(méi)有half消息這個(gè)流程，那我們通常是會(huì)在訂單系統(tǒng)中先完成下單，再發(fā)送消息給MQ。這時(shí)候?qū)懭胂⒌組Q如果失敗就會(huì)非常尷尬了。而half消息如果寫(xiě)入失敗，我們就可以認(rèn)為MQ的服務(wù)是有問(wèn)題的，這時(shí)，就不能通知下游服務(wù)了。我們可以在下單時(shí)給訂單一個(gè)狀態(tài)標(biāo)記，然后等待MQ服務(wù)正常后再進(jìn)行補(bǔ)償操作，等MQ服務(wù)正常后重新下單通知下游服務(wù)。

（3）.訂單系統(tǒng)寫(xiě)數(shù)據(jù)庫(kù)失敗了怎么辦？ 這個(gè)問(wèn)題我們同樣比較下沒(méi)有使用事務(wù)消息機(jī)制時(shí)會(huì)怎么辦？如果沒(méi)有使用事務(wù)消息，我們只能判斷下單失敗，拋出了異常，那就不往MQ發(fā)消息了，這樣至少保證不會(huì)對(duì)下游服務(wù)進(jìn)行錯(cuò)誤的通知。但是這樣的話(huà)，如果過(guò)一段時(shí)間數(shù)據(jù)庫(kù)恢復(fù)過(guò)來(lái)了，這個(gè)消息就無(wú)法再次發(fā)送了。當(dāng)然，也可以設(shè)計(jì)另外的補(bǔ)償機(jī)制，例如將訂單數(shù)據(jù)緩存起來(lái)，再啟動(dòng)一個(gè)線(xiàn)程定時(shí)嘗試往數(shù)據(jù)庫(kù)寫(xiě)。而如果使用事務(wù)消息機(jī)制，就可以有一種更優(yōu)雅的方案。 如果下單時(shí)，寫(xiě)數(shù)據(jù)庫(kù)失敗(可能是數(shù)據(jù)庫(kù)崩了，需要等一段時(shí)間才能恢復(fù))。那我們可以另外找個(gè)地方把訂單消息先緩存起來(lái)(Redis、文本或者其他方式)，然后給RocketMQ返回一個(gè)UNKNOWN狀態(tài)。這樣RocketMQ就會(huì)過(guò)一段時(shí)間來(lái)回查事務(wù)狀態(tài)。我們就可以在回查事務(wù)狀態(tài)時(shí)再?lài)L試把訂單數(shù)據(jù)寫(xiě)入數(shù)據(jù)庫(kù)，如果數(shù)據(jù)庫(kù)這時(shí)候已經(jīng)恢復(fù)了，那就能完整正常的下單，再繼續(xù)后面的業(yè)務(wù)。這樣這個(gè)訂單的消息就不會(huì)因?yàn)閿?shù)據(jù)庫(kù)臨時(shí)崩了而丟失。

（4）.half消息寫(xiě)入成功后RocketMQ掛了怎么辦？

我們需要注意下，在事務(wù)消息的處理機(jī)制中，未知狀態(tài)的事務(wù)狀態(tài)回查是由RocketMQ的Broker主動(dòng)發(fā)起的。也就是說(shuō)如果出現(xiàn)了這種情況，那RocketMQ就不會(huì)回調(diào)到事務(wù)消息中回查事務(wù)狀態(tài)的服務(wù)。這時(shí)，我們就可以將訂單一直標(biāo)記為"新下單"的狀態(tài)。而等RocketMQ恢復(fù)后，只要存儲(chǔ)的消息沒(méi)有丟失，RocketMQ就會(huì)再次繼續(xù)狀態(tài)回查的流程。

（5）.下單成功后如何優(yōu)雅的等待支付成功？

在訂單場(chǎng)景下，通常會(huì)要求下單完成后，客戶(hù)在一定時(shí)間內(nèi)，例如10分鐘，內(nèi)完成訂單支付，支付完成后才會(huì)通知下游服務(wù)進(jìn)行進(jìn)一步的營(yíng)銷(xiāo)補(bǔ)償。 如果不用事務(wù)消息，那通常會(huì)怎么辦？ 最簡(jiǎn)單的方式是啟動(dòng)一個(gè)定時(shí)任務(wù)，每隔一段時(shí)間掃描訂單表，比對(duì)未支付的訂單的下單時(shí)間，將超過(guò)時(shí)間的訂單回收。這種方式顯然是有很大問(wèn)題的，需要定時(shí)掃描很龐大的一個(gè)訂單信息，這對(duì)系統(tǒng)是個(gè)不小的壓力。 那更進(jìn)一步的方案是什么呢？是不是就可以使用RocketMQ提供的延遲消息機(jī)制。往MQ發(fā)一個(gè)延遲1分鐘的消息，消費(fèi)到這個(gè)消息后去檢查訂單的支付狀態(tài)，如果訂單已經(jīng)支付，就往下游發(fā)送下單的通知。而如果沒(méi)有支付，就再發(fā)一個(gè)延遲1分鐘的消息。最終在第十個(gè)消息時(shí)把訂單回收。這個(gè)方案就不用對(duì)全部的訂單表進(jìn)行掃描，而只需要每次處理一個(gè)單獨(dú)的訂單消息。 那如果使用上了事務(wù)消息呢？我們就可以用事務(wù)消息的狀態(tài)回查機(jī)制來(lái)替代定時(shí)的任務(wù)。在下單時(shí)，給Broker返回一個(gè)UNKNOWN的未知狀態(tài)。而在狀態(tài)回查的方法中去查詢(xún)訂單的支付狀態(tài)。這樣整個(gè)業(yè)務(wù)邏輯就會(huì)簡(jiǎn)單很多。我們只需要配置RocketMQ中的事務(wù)消息回查次數(shù)(默認(rèn)15次)和事務(wù)回查間隔時(shí)間(messageDelayLevel)，就可以更優(yōu)雅的完成這個(gè)支付狀態(tài)檢查的需求。

（6）、事務(wù)消息機(jī)制的作用

整體來(lái)說(shuō)，在訂單這個(gè)場(chǎng)景下，消息不丟失的問(wèn)題實(shí)際上就還是轉(zhuǎn)化成了下單這個(gè)業(yè)務(wù)與下游服務(wù)的業(yè)務(wù)的分布式事務(wù)一致性問(wèn)題。而事務(wù)一致性問(wèn)題一直以來(lái)都是一個(gè)非常復(fù)雜的問(wèn)題。而RocketMQ的事務(wù)消息機(jī)制，實(shí)際上只保證了整個(gè)事務(wù)消息的一半，他保證的是訂單系統(tǒng)下單和發(fā)消息這兩個(gè)事件的事務(wù)一致性，而對(duì)下游服務(wù)的事務(wù)并沒(méi)有保證。但是即便如此，也是分布式事務(wù)的一個(gè)很好的降級(jí)方案。目前來(lái)看，也是業(yè)內(nèi)最好的降級(jí)方案。

2、RocketMQ配置同步刷盤(pán)+Dledger主從架構(gòu)保證MQ自身不會(huì)丟消息

（1）、同步刷盤(pán) 這個(gè)從我們之前的分析，就很好理解了。我們可以簡(jiǎn)單的把RocketMQ的刷盤(pán)方式 flushDiskType配置成同步刷盤(pán)就可以保證消息在刷盤(pán)過(guò)程中不會(huì)丟失了。

（2）、Dledger的文件同步

在使用Dledger技術(shù)搭建的RocketMQ集群中，Dledger會(huì)通過(guò)兩階段提交的方式保證文件在主從之間成功同步。 簡(jiǎn)單來(lái)說(shuō)，數(shù)據(jù)同步會(huì)通過(guò)兩個(gè)階段，一個(gè)是uncommitted階段，一個(gè)是commited階段。 Leader Broker上的Dledger收到一條數(shù)據(jù)后，會(huì)標(biāo)記為uncommitted狀態(tài)，然后他通過(guò)自己的DledgerServer組件把這個(gè)uncommitted數(shù)據(jù)發(fā)給Follower Broker的DledgerServer組件。 接著Follower Broker的DledgerServer收到uncommitted消息之后，必須返回一個(gè)ack給Leader Broker的Dledger。然后如果Leader Broker收到超過(guò)半數(shù)的Follower Broker返回的ack之后，就會(huì)把消息標(biāo)記為committed狀態(tài)。 再接下來(lái)， Leader Broker上的DledgerServer就會(huì)發(fā)送committed消息給Follower Broker上的DledgerServer，讓他們把消息也標(biāo)記為committed狀態(tài)。這樣，就基于Raft協(xié)議完成了兩階段的數(shù)據(jù)同步。

3、消費(fèi)者端不要使用異步消費(fèi)機(jī)制

正常情況下，消費(fèi)者端都是需要先處理本地事務(wù)，然后再給MQ一個(gè)ACK響應(yīng)，這時(shí)MQ就會(huì)修改Offset，將消息標(biāo)記為已消費(fèi)，從而不再往其他消費(fèi)者推送消息。所以在Broker的這種重新推送機(jī)制下，消息是不會(huì)在傳輸過(guò)程中丟失的。但是也會(huì)有下面這種情況會(huì)造成服務(wù)端消息丟失：

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {new Thread(){public void run(){//處理業(yè)務(wù)邏輯System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);}};return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}
});

這種異步消費(fèi)的方式，就有可能造成消息狀態(tài)返回后消費(fèi)者本地業(yè)務(wù)邏輯處理失敗造成消息丟失的可能。

4、RocketMQ特有的問(wèn)題，NameServer掛了如何保證消息不丟失？

NameServer在RocketMQ中，是扮演的一個(gè)路由中心的角色，提供到Broker的路由功能。但是其實(shí)路由中心這樣的功能，在所有的MQ中都是需要的。kafka是用zookeeper和一個(gè)作為Controller的Broker一起來(lái)提供路由服務(wù)，整個(gè)功能是相當(dāng)復(fù)雜糾結(jié)的。而RabbitMQ是由每一個(gè)Broker來(lái)提供路由服務(wù)。而只有RocketMQ把這個(gè)路由中心單獨(dú)抽取了出來(lái)，并獨(dú)立部署。 這個(gè)NameServer之前都了解過(guò)，集群中任意多的節(jié)點(diǎn)掛掉，都不會(huì)影響他提供的路由功能。那如果集群中所有的NameServer節(jié)點(diǎn)都掛了呢？ 有很多人就會(huì)認(rèn)為在生產(chǎn)者和消費(fèi)者中都會(huì)有全部路由信息的緩存副本，那整個(gè)服務(wù)可以正常工作一段時(shí)間。其實(shí)這個(gè)問(wèn)題大家可以做一下實(shí)驗(yàn)，當(dāng)NameServer全部掛了后，生產(chǎn)者和消費(fèi)者是立即就無(wú)法工作了的。至于為什么，可以回顧一下我們之前的源碼課程去源碼中找找答案。 那再回到我們的消息不丟失的問(wèn)題，在這種情況下，RocketMQ相當(dāng)于整個(gè)服務(wù)都不可用了，那他本身肯定無(wú)法給我們保證消息不丟失了。我們只能自己設(shè)計(jì)一個(gè)降級(jí)方案來(lái)處理這個(gè)問(wèn)題了。例如在訂單系統(tǒng)中，如果多次嘗試發(fā)送RocketMQ不成功，那就只能另外找給地方(Redis、文件或者內(nèi)存等)把訂單消息緩存下來(lái)，然后起一個(gè)線(xiàn)程定時(shí)的掃描這些失敗的訂單消息，嘗試往RocketMQ發(fā)送。這樣等RocketMQ的服務(wù)恢復(fù)過(guò)來(lái)后，就能第一時(shí)間把這些消息重新發(fā)送出去。整個(gè)這套降級(jí)的機(jī)制，在大型互聯(lián)網(wǎng)項(xiàng)目中，都是必須要有的。

完整分析過(guò)后，整個(gè)RocketMQ消息零丟失的方案其實(shí)挺簡(jiǎn)單

• 生產(chǎn)者使用事務(wù)消息機(jī)制。
• Broker配置同步刷盤(pán)+Dledger主從架構(gòu)
• 消費(fèi)者不要使用異步消費(fèi)。
• 整個(gè)MQ掛了之后準(zhǔn)備降級(jí)方案

使用RocketMQ如何快速處理積壓消息？

1、如何確定RocketMQ有大量的消息積壓？

在正常情況下，使用MQ都會(huì)要盡量保證他的消息生產(chǎn)速度和消費(fèi)速度整體上是平衡的，但是如果部分消費(fèi)者系統(tǒng)出現(xiàn)故障，就會(huì)造成大量的消息積累。這類(lèi)問(wèn)題通常在實(shí)際工作中會(huì)出現(xiàn)得比較隱蔽。例如某一天一個(gè)數(shù)據(jù)庫(kù)突然掛了，大家大概率就會(huì)集中處理數(shù)據(jù)庫(kù)的問(wèn)題。等好不容易把數(shù)據(jù)庫(kù)恢復(fù)過(guò)來(lái)了，這時(shí)基于這個(gè)數(shù)據(jù)庫(kù)服務(wù)的消費(fèi)者程序就會(huì)積累大量的消息?；蛘呔W(wǎng)絡(luò)波動(dòng)等情況，也會(huì)導(dǎo)致消息大量的積累。這在一些大型的互聯(lián)網(wǎng)項(xiàng)目中，消息積壓的速度是相當(dāng)恐怖的。所以消息積壓是個(gè)需要時(shí)時(shí)關(guān)注的問(wèn)題。 對(duì)于消息積壓，如果是RocketMQ或者kafka還好，他們的消息積壓不會(huì)對(duì)性能造成很大的影響。而如果是RabbitMQ的話(huà)，那就慘了，大量的消息積壓可以瞬間造成性能直線(xiàn)下滑。 對(duì)于RocketMQ來(lái)說(shuō)，有個(gè)最簡(jiǎn)單的方式來(lái)確定消息是否有積壓。那就是使用web控制臺(tái)，就能直接看到消息的積壓情況。 在Web控制臺(tái)的主題頁(yè)面，可以通過(guò) Consumer管理 按鈕實(shí)時(shí)看到消息的積壓情況。

另外，也可以通過(guò)mqadmin指令在后臺(tái)檢查各個(gè)Topic的消息延遲情況。 還有RocketMQ也會(huì)在他的 ${storePathRootDir}/config 目錄下落地一系列的json文件，也可以用來(lái)跟蹤消息積壓情況。

2、如何處理大量積壓的消息？

其實(shí)我們回顧下RocketMQ的負(fù)載均衡的內(nèi)容就不難想到解決方案。 如果Topic下的MessageQueue配置得是足夠多的，那每個(gè)Consumer實(shí)際上會(huì)分配多個(gè)MessageQueue來(lái)進(jìn)行消費(fèi)。這個(gè)時(shí)候，就可以簡(jiǎn)單的通過(guò)增加Consumer的服務(wù)節(jié)點(diǎn)數(shù)量來(lái)加快消息的消費(fèi)，等積壓消息消費(fèi)完了，再恢復(fù)成正常情況。最極限的情況是把Consumer的節(jié)點(diǎn)個(gè)數(shù)設(shè)置成跟MessageQueue的個(gè)數(shù)相同。但是如果此時(shí)再繼續(xù)增加Consumer的服務(wù)節(jié)點(diǎn)就沒(méi)有用了。 而如果Topic下的MessageQueue配置得不夠多的話(huà)，那就不能用上面這種增加Consumer節(jié)點(diǎn)個(gè)數(shù)的方法了。這時(shí)怎么辦呢？ 這時(shí)如果要快速處理積壓的消息，可以創(chuàng)建一個(gè)新的Topic，配置足夠多的MessageQueue。然后把所有消費(fèi)者節(jié)點(diǎn)的目標(biāo)Topic轉(zhuǎn)向新的Topic，并緊急上線(xiàn)一組新的消費(fèi)者，只負(fù)責(zé)消費(fèi)舊Topic中的消息，并轉(zhuǎn)儲(chǔ)到新的Topic中，這個(gè)速度是可以很快的。然后在新的Topic上，就可以通過(guò)增加消費(fèi)者個(gè)數(shù)來(lái)提高消費(fèi)速度了。之后再根據(jù)情況恢復(fù)成正常情況。 在官網(wǎng)中，還分析了一個(gè)特殊的情況。就是如果RocketMQ原本是采用的普通方式搭建主從架構(gòu)，而現(xiàn)在想要中途改為使用Dledger高可用集群，這時(shí)候如果不想歷史消息丟失，就需要先將消息進(jìn)行對(duì)齊，也就是要消費(fèi)者把所有的消息都消費(fèi)完，再來(lái)切換主從架構(gòu)。因?yàn)镈ledger集群會(huì)接管RocketMQ原有的CommitLog日志，所以切換主從架構(gòu)時(shí)，如果有消息沒(méi)有消費(fèi)完，這些消息是存在舊的CommitLog中的，就無(wú)法再進(jìn)行消費(fèi)了。這個(gè)場(chǎng)景下也是需要盡快的處理掉積壓的消息。

RocketMQ的消息軌跡

RocketMQ默認(rèn)提供了消息軌跡的功能：

另外，也支持客戶(hù)端自定義軌跡數(shù)據(jù)存儲(chǔ)的Topic。 在客戶(hù)端的兩個(gè)核心對(duì)象 DefaultMQProducer和DefaultMQPushConsumer，他們的構(gòu)造函數(shù)中，都有兩個(gè)可選的參數(shù)來(lái)打開(kāi)消息軌跡存儲(chǔ)

• enableMsgTrace：是否打開(kāi)消息軌跡。默認(rèn)是false。

• customizedTraceTopic：配置將消息軌跡數(shù)據(jù)存儲(chǔ)到用戶(hù)指定的Topic 。

RocketMQ 如何處理流量削峰

連接地址： [(23條消息) RocketMQ實(shí)現(xiàn)流量削峰_mq削峰_ClareTung的博客-CSDN博客](https://blog.csdn.net/qq_36135928/article/details/121146631?ops_request_misc=&request_id=&biz_id=102&utm_term=RocketMQ 如何處理流量肖峰&utm_medium=distribute.pc_search_result.none-task-blog-2_allsobaiduweb~default-0-121146631.142^v80insert_down1,201^v4add_ask,239^v2insert_chatgpt&spm=1018.2226.3001.4187)

實(shí)現(xiàn)的思路：

總體思路

• http請(qǐng)求到gateway網(wǎng)關(guān)
• gateway通過(guò)RouteLocator配置路由信息轉(zhuǎn)發(fā)到mq服務(wù)
• mq服務(wù)接收到數(shù)據(jù)后，將數(shù)據(jù)發(fā)送到對(duì)應(yīng)的topic
• mq消費(fèi)者監(jiān)聽(tīng)消息，拉取消息，進(jìn)行業(yè)務(wù)處理，處理完后將msgId作為key，結(jié)果作為value存入redis 中
• mq服務(wù)發(fā)送消息的線(xiàn)程發(fā)送完消息后掛起，每隔一秒根據(jù)msgId去redis查詢(xún)消費(fèi)結(jié)果

使用RocketMQ如何保證消息順序

為什么要保證消息的有序：例如如果我們有個(gè)大數(shù)據(jù)系統(tǒng)，需要對(duì)業(yè)務(wù)系統(tǒng)的日志進(jìn)行收集分析，這時(shí)候?yàn)榱藴p少對(duì)業(yè)務(wù)系統(tǒng)的影響，通常都會(huì)通過(guò)MQ來(lái)做消息中轉(zhuǎn)。而這時(shí)候，對(duì)消息的順序就有一定的要求了。

MQ的順序問(wèn)題分為全局有序和局部有序：
全局有序：整個(gè)MQ系統(tǒng)的所有消息嚴(yán)格按照隊(duì)列先入先出順序進(jìn)行消費(fèi)。
局部有序：只保證一部分關(guān)鍵消息的消費(fèi)順序。