列Kafka、RabbitMQ、RocketMQ選型對(duì)比與實(shí)戰(zhàn)指南)
先說(shuō)個(gè)背景。做后端這幾年消息隊(duì)列基本是繞不開(kāi)的中間件。不管是電商的訂單流轉(zhuǎn)、日志收集、異步通知還是高并發(fā)下的削峰填谷你總會(huì)遇到一個(gè)場(chǎng)景需要引入消息隊(duì)列。市面上一堆MQ但真正經(jīng)常被提起、被面試官追問(wèn)、被生產(chǎn)環(huán)境大規(guī)模使用的無(wú)非就是Kafka、RabbitMQ和RocketMQ這三個(gè)。我這些年三個(gè)隊(duì)列都用過(guò)踩過(guò)不少坑也總結(jié)出一些實(shí)戰(zhàn)經(jīng)驗(yàn)這篇就一次性把它們從選型、部署、代碼接入到問(wèn)題排查完整串一遍。這篇內(nèi)容不會(huì)照著官方文檔念而是從實(shí)際使用者的視角出發(fā)講清楚每個(gè)隊(duì)列的核心原理、安裝關(guān)鍵點(diǎn)和代碼落地的真實(shí)細(xì)節(jié)。不管你是剛接觸消息隊(duì)列的新人還是已經(jīng)用了一段時(shí)間、想系統(tǒng)補(bǔ)一遍的開(kāi)發(fā)者都能從中找到可直接參考的東西。文章比較長(zhǎng)建議先收藏等真正用的時(shí)候再回來(lái)翻。1. 提筆之前消息隊(duì)列到底解決什么問(wèn)題三大隊(duì)列該怎么選1.1 消息隊(duì)列的核心價(jià)值不是隊(duì)列而是解耦很多人理解消息隊(duì)列第一反應(yīng)就是先進(jìn)先出的數(shù)據(jù)結(jié)構(gòu)。這個(gè)理解沒(méi)錯(cuò)但在實(shí)際工程里消息隊(duì)列的核心價(jià)值根本不是隊(duì)列本身而是它帶來(lái)的三個(gè)能力異步、削峰、解耦。我再拆細(xì)一點(diǎn)。異步用戶下單后系統(tǒng)需要發(fā)短信、發(fā)優(yōu)惠券、更新積分。如果這些步驟全部串行執(zhí)行一次請(qǐng)求可能要等幾百毫秒甚至更久。把非核心操作丟進(jìn)隊(duì)列主鏈路只處理扣庫(kù)存、生成訂單響應(yīng)時(shí)間直接從秒級(jí)降到毫秒級(jí)。削峰秒殺場(chǎng)景下瞬間涌入十萬(wàn)請(qǐng)求數(shù)據(jù)庫(kù)根本扛不住。讓請(qǐng)求先進(jìn)隊(duì)列后端按自身處理能力慢慢消費(fèi)系統(tǒng)就不會(huì)被流量打垮。解耦訂單服務(wù)不需要知道庫(kù)存服務(wù)、物流服務(wù)的具體實(shí)現(xiàn)只要往隊(duì)列里發(fā)一條訂單創(chuàng)建成功的消息下游服務(wù)各自訂閱自己關(guān)心的內(nèi)容互不干擾。這也是為什么現(xiàn)在的面試題里消息隊(duì)列幾乎是必問(wèn)項(xiàng)。面試官想知道的不只是你會(huì)不會(huì)調(diào)API而是你有沒(méi)有真正理解它解決什么問(wèn)題、在什么場(chǎng)景下選哪個(gè)。1.2 三個(gè)隊(duì)列的定位差異與選型邏輯Kafka、RabbitMQ、RocketMQ雖然都叫消息隊(duì)列但出生背景完全不同這也決定了它們的主戰(zhàn)場(chǎng)不同。Kafka出生在LinkedIn最初是為了處理海量日志流而設(shè)計(jì)的。它的架構(gòu)天然適合大規(guī)模數(shù)據(jù)管道、日志采集、流式計(jì)算這類場(chǎng)景吞吐量極高但功能相對(duì)寡淡——延時(shí)消息、死信隊(duì)列這些高級(jí)特性原生支持都不太好得靠外部或自己實(shí)現(xiàn)。RabbitMQ是Erlang寫的老牌消息中間件功能非常豐富路由模式靈活文檔生態(tài)完善適合企業(yè)內(nèi)部業(yè)務(wù)系統(tǒng)之間做消息通信。RocketMQ是阿里開(kāi)源的消息中間件人家是從電商業(yè)務(wù)里長(zhǎng)出來(lái)的所以天然補(bǔ)齊了阿里巴巴在交易鏈路上的各種需求事務(wù)消息、延時(shí)消息、消息重試、死信隊(duì)列甚至消息軌跡追蹤都有。簡(jiǎn)單總結(jié)選型邏輯你們做的是大數(shù)據(jù)管道、日志采集、流計(jì)算那一路選Kafka業(yè)務(wù)系統(tǒng)內(nèi)部需要靈活的路由、優(yōu)先級(jí)、RPC調(diào)用通知選RabbitMQ業(yè)務(wù)量很大并且深度使用Java需要事務(wù)消息和可靠投遞選RocketMQ。這不是絕對(duì)的但九成場(chǎng)景這么選都不會(huì)錯(cuò)。2. RabbitMQ功能最全的業(yè)務(wù)消息中間件從安裝到生產(chǎn)落地2.1 安裝部署Windows、Linux和Docker三種方式對(duì)比RabbitMQ在Windows上的安裝以前是不少人入門的第一道坎。它依賴Erlang環(huán)境需要先下載Erlang的Windows安裝包再下載RabbitMQ Server的Windows安裝包而且版本要對(duì)應(yīng)。網(wǎng)上關(guān)于RabbitMQ下載和安裝的教程很多但我見(jiàn)過(guò)太多人卡在Erlang版本和RabbitMQ版本不匹配這個(gè)問(wèn)題上。官網(wǎng)上每個(gè)RabbitMQ版本都會(huì)標(biāo)注對(duì)應(yīng)的Erlang版本區(qū)間安裝前務(wù)必核對(duì)。Windows下安裝完還需要手動(dòng)激活管理插件命令行執(zhí)行rabbitmq-plugins enable rabbitmq_management然后重啟RabbitMQ服務(wù)瀏覽器訪問(wèn)http://localhost:15672 用默認(rèn)賬號(hào)guest/guest登錄。這里有個(gè)坑默認(rèn)的guest賬號(hào)只能在localhost訪問(wèn)如果你用IP地址訪問(wèn)管理臺(tái)會(huì)被拒絕。要么用localhost訪問(wèn)要么新建一個(gè)用戶并賦予權(quán)限。Linux下安裝稍微順暢一些因?yàn)楣俜教峁┝薬pt和yum源。以Ubuntu為例sudo apt-get install -y rabbitmq-server sudo systemctl enable rabbitmq-server sudo systemctl start rabbitmq-server sudo rabbitmq-plugins enable rabbitmq_management但最推薦的方式還是Docker尤其適合本地開(kāi)發(fā)和測(cè)試環(huán)境。一條命令搞定還不用操心底層依賴docker run -d --name rabbitmq \ -p 5672:5672 -p 15672:15672 \ -e RABBITMQ_DEFAULT_USERadmin \ -e RABBITMQ_DEFAULT_PASSadmin123 \ rabbitmq:3.12-management這里要注意鏡像分為兩個(gè)版本rabbitmq版本號(hào)不帶management和rabbitmq版本號(hào)-management帶管理臺(tái)插件。本地調(diào)試建議直接用management版本省得手動(dòng)裝插件。2.2 核心概念與Java客戶端接入RabbitMQ的核心模型是生產(chǎn)者Producer把消息發(fā)給交換機(jī)Exchange交換機(jī)根據(jù)路由規(guī)則把消息投遞到綁定的隊(duì)列Queue消費(fèi)者Consumer從隊(duì)列拉取消息處理。很多人第一反應(yīng)是為什么不直接發(fā)到隊(duì)列原因在于交換機(jī)提供了靈活的路由能力。交換機(jī)有四種類型Direct點(diǎn)對(duì)點(diǎn)精確匹配、Fanout廣播給所有綁定的隊(duì)列、Topic通配符匹配路由鍵、Headers基于消息頭的匹配用得少。實(shí)際工程中Topic交換機(jī)用得最多因?yàn)樗钠ヅ湟?guī)則靈活。比如訂單系統(tǒng)發(fā)一條消息路由鍵是order.created物流服務(wù)綁定order.*積分服務(wù)綁定order.created兩個(gè)服務(wù)都能收到消息非常方便。Java客戶端接入的代碼核心就三塊。第一塊是創(chuàng)建連接工廠ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); factory.setUsername(admin); factory.setPassword(admin123); Connection connection factory.newConnection(); Channel channel connection.createChannel();第二塊是聲明隊(duì)列和交換機(jī)并做綁定channel.exchangeDeclare(order.exchange, topic, true); channel.queueDeclare(order.queue, true, false, false, null); channel.queueBind(order.queue, order.exchange, order.#);第三塊是發(fā)送消息和接收消息。發(fā)送String message 訂單創(chuàng)建成功; channel.basicPublish(order.exchange, order.created, null, message.getBytes());接收channel.basicConsume(order.queue, false, (consumerTag, delivery) - { String msg new String(delivery.getBody(), UTF-8); System.out.println(收到消息: msg); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); }, consumerTag - {});注意basicConsume的第二個(gè)參數(shù)是autoAck工作中我建議一律顯式設(shè)為false手動(dòng)調(diào)用basicAck確認(rèn)消息。原因很簡(jiǎn)單如果設(shè)置了自動(dòng)確認(rèn)消費(fèi)者收到消息但還沒(méi)來(lái)得及處理就掛了消息就丟了手動(dòng)確認(rèn)可以確保消息被完整處理后才會(huì)從隊(duì)列移除。2.3 消息可靠性生產(chǎn)端確認(rèn)與消費(fèi)端手動(dòng)Ack消息隊(duì)列最讓人頭疼的問(wèn)題就是消息丟失。RabbitMQ的消息丟失可能發(fā)生在三個(gè)環(huán)節(jié)生產(chǎn)者發(fā)送時(shí)、隊(duì)列存儲(chǔ)時(shí)、消費(fèi)者消費(fèi)時(shí)。生產(chǎn)端要開(kāi)啟生產(chǎn)確認(rèn)模式。在channel上調(diào)用confirmSelect()方法發(fā)布消息后等待broker的確認(rèn)回調(diào)不確認(rèn)就重發(fā)。這是第一層保障。隊(duì)列持久化第二層保障聲明隊(duì)列時(shí)把durable參數(shù)設(shè)為true這樣RabbitMQ重啟后隊(duì)列和消息還在。消費(fèi)端手動(dòng)Ack第三層保障前面已經(jīng)提到過(guò)。這三層都做好了消息才能做到不丟失。很多人配置完不管不顧直到某天凌晨收到告警郵件才發(fā)現(xiàn)消息丟了一查結(jié)果就是消費(fèi)者沒(méi)做手動(dòng)確認(rèn)。這類問(wèn)題我在生產(chǎn)環(huán)境見(jiàn)過(guò)太多次。2.4 高頻問(wèn)題延時(shí)隊(duì)列、死信隊(duì)列與重復(fù)消費(fèi)RabbitMQ原生的延時(shí)隊(duì)列是通過(guò)死信交換機(jī)DLX機(jī)制實(shí)現(xiàn)的。思路是給隊(duì)列設(shè)置一個(gè)TTL消息過(guò)期后會(huì)被轉(zhuǎn)投到綁定的死信交換機(jī)再由死信交換機(jī)路由到真正的處理隊(duì)列。這個(gè)方案雖然繞但生產(chǎn)環(huán)境很常見(jiàn)。比如電商下單后30分鐘未支付自動(dòng)關(guān)閉訂單就是用延時(shí)隊(duì)列實(shí)現(xiàn)的。所以RabbitMQ面試題里死信隊(duì)列是絕對(duì)的高頻考點(diǎn)。實(shí)現(xiàn)要點(diǎn)是創(chuàng)建兩個(gè)隊(duì)列和一個(gè)交換機(jī)一個(gè)普通隊(duì)列帶DLX配置和TTL一個(gè)實(shí)際處理隊(duì)列綁定死信交換機(jī)。重復(fù)消費(fèi)問(wèn)題則是所有消息隊(duì)列的通病。根本原因在于分布式網(wǎng)絡(luò)環(huán)境下消息可能因?yàn)榫W(wǎng)絡(luò)抖動(dòng)導(dǎo)致消費(fèi)者已經(jīng)處理完但Ack沒(méi)送達(dá)brokerbroker會(huì)重發(fā)消息。解決思路也比較統(tǒng)一消費(fèi)者側(cè)做冪等處理。常見(jiàn)方案有用redis setnx做冪等鍵、數(shù)據(jù)庫(kù)唯一約束、業(yè)務(wù)表里加一個(gè)處理狀態(tài)字段。我在項(xiàng)目里用的最多的是業(yè)務(wù)鍵Redis的方式消息到達(dá)先檢查Redis里是否已存在該消息ID存在則直接返回不存在則處理并寫入Redis。3. Kafka高吞吐的日志與流數(shù)據(jù)處理利器原理和集群安裝一次說(shuō)清3.1 Kafka的架構(gòu)與工作原理先弄懂Kafka的基本模型BrokerKafka服務(wù)器節(jié)點(diǎn)、Topic消息分類、PartitionTopic的分區(qū)、Producer生產(chǎn)者、Consumer消費(fèi)者和Consumer Group消費(fèi)者組。Kafka最大的特點(diǎn)是分區(qū)并行這也是它吞吐量遠(yuǎn)超其他MQ的關(guān)鍵。一個(gè)Topic可以分成多個(gè)Partition每個(gè)Partition是獨(dú)立的日志文件讀寫互不干擾。消費(fèi)者組里的每個(gè)消費(fèi)者負(fù)責(zé)部分分區(qū)實(shí)現(xiàn)水平擴(kuò)展。Kafka的消息在Partition內(nèi)是有序的不同Partition之間無(wú)順序保證。Kafka的存儲(chǔ)機(jī)制也很特別。消息寫入后不是立即刪除而是保存在磁盤里基于順序?qū)懞晚?yè)緩存技術(shù)保證高吞吐。消息的消費(fèi)進(jìn)度由消費(fèi)者自己管理——記錄當(dāng)前消費(fèi)到哪個(gè)Offset。這也解釋了為什么Kafka的消費(fèi)者可以做到隨時(shí)回溯你可以把Offset重置到昨天重新消費(fèi)一遍。3.2 單機(jī)與集群安裝Kafka本身依賴ZooKeeper新版雖然引入了KRaft模式但大多數(shù)生產(chǎn)環(huán)境還在用ZooKeeper安裝Kafka基本離不開(kāi)ZooKeeper。這也是Kafka安裝鏈路比其他MQ繁重的原因。單機(jī)安裝比較簡(jiǎn)單這里是Linux環(huán)境操作# 下載解壓kafka wget https://downloads.apache.org/kafka/3.4.0/kafka_2.13-3.4.0.tgz tar -zxvf kafka_2.13-3.4.0.tgz cd kafka_2.13-3.4.0 # 啟動(dòng)ZooKeeper bin/zookeeper-server-start.sh config/zookeeper.properties # 啟動(dòng)Kafka bin/kafka-server-start.sh config/server.properties 集群安裝本質(zhì)上就是多節(jié)點(diǎn)配置。每個(gè)Broker要修改config/server.properties里的幾個(gè)關(guān)鍵參數(shù)broker.id集群內(nèi)唯一、listeners監(jiān)聽(tīng)地址、log.dirs日志目錄、zookeeper.connect指向ZooKeeper集群。三節(jié)點(diǎn)的集群配置大致是這樣node1: broker.id0listenersPLAINTEXT://192.168.1.10:9092node2: broker.id1listenersPLAINTEXT://192.168.1.11:9092node3: broker.id2listenersPLAINTEXT://192.168.1.12:9092ZooKeeper地址都寫成zookeeper集群的地址列表比如192.168.1.1:2181,192.168.1.2:2181,192.168.1.3:2181。如果是離線環(huán)境Kafka集群離線安裝也不復(fù)雜——就是把已經(jīng)下載好的tar包和依賴包通過(guò)內(nèi)網(wǎng)傳輸?shù)侥繕?biāo)機(jī)器安裝步驟和在線一致。關(guān)鍵在于三臺(tái)機(jī)器的時(shí)鐘要同步建議配置NTP。集群配置完成后用kafka-topics.sh創(chuàng)建topic時(shí)指定replication-factor數(shù)據(jù)才會(huì)在多個(gè)broker間復(fù)制。3.3 生產(chǎn)與消費(fèi)命令實(shí)操Kafka的命令行是排查問(wèn)題的利器熟練掌握能省很多時(shí)間。創(chuàng)建主題bin/kafka-topics.sh --create \ --bootstrap-server localhost:9092 \ --replication-factor 1 --partitions 3 \ --topic order-topic生產(chǎn)消息測(cè)試bin/kafka-console-producer.sh --broker-list localhost:9092 --topic order-topic消費(fèi)消息bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \ --topic order-topic --from-beginning這條消費(fèi)命令有一個(gè)參數(shù)經(jīng)常被忽略--from-beginning。它的含義是從最早的Offset開(kāi)始消費(fèi)。如果你想讓消費(fèi)者從指定時(shí)間點(diǎn)開(kāi)始消費(fèi)可以加--offset 指定Offset位置或者用--partition加--offset組合。熱搜詞里提到的kafka消費(fèi)命令指定消費(fèi)時(shí)間實(shí)際做法是用kafka-consumer-groups.sh重置偏移量bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \ --group my-group --topic order-topic \ --reset-offsets --to-datetime 2024-01-01T00:00:00.000 --execute這條命令很實(shí)用當(dāng)年排查線上數(shù)據(jù)異常時(shí)就是用這個(gè)把消費(fèi)者組的Offset重置到出問(wèn)題之前的某個(gè)時(shí)間點(diǎn)讓消費(fèi)者重新消費(fèi)那段歷史消息。3.4 Kafka可視化工具與消息延遲排查Kafka是出了名的難以可視化的中間件不像RabbitMQ自帶一個(gè)漂亮的管理臺(tái)。很多剛接觸Kafka的人都會(huì)搜kafka可視化工具、kafka連接工具、kafka圖形界面。這里我推薦兩個(gè)Kafka Tool現(xiàn)在叫Offset Explorer和Kafka UI。前者是桌面客戶端圖形化查看Topic、Partition、Consumer Group的Offset情況適合日常排查后者是Web界面支持集群管理、消息查看和動(dòng)態(tài)創(chuàng)建Topic適合團(tuán)隊(duì)共用。關(guān)于Kafka消息延遲高的問(wèn)題幾乎每個(gè)團(tuán)隊(duì)都會(huì)遇到。我排查這類問(wèn)題的經(jīng)驗(yàn)是按鏈路逐個(gè)檢查先看生產(chǎn)者端發(fā)送耗時(shí)是不是發(fā)送時(shí)同步等待過(guò)久如果批量發(fā)送參數(shù)設(shè)置不合理會(huì)影響性能。再看消費(fèi)者端處理速度是不是單條消息處理邏輯太慢或者消費(fèi)者線程數(shù)太低。重點(diǎn)看兩個(gè)指標(biāo)Consumer Group的Lag積壓量和每個(gè)分區(qū)的處理耗時(shí)。如果Lag持續(xù)增長(zhǎng)說(shuō)明消費(fèi)速度趕不上生產(chǎn)速度。處理人最多的方案是加消費(fèi)者實(shí)例增加單個(gè)消費(fèi)者的線程數(shù)或者優(yōu)化消費(fèi)邏輯的DB訪問(wèn)比如把單條插入改成批量插入。還有一點(diǎn)很容易被忽略如果某些消息處理邏輯里有外部API調(diào)用而外部服務(wù)響應(yīng)慢會(huì)直接拖垮整個(gè)消費(fèi)者的消費(fèi)速度。這種情況建議把外部調(diào)用改成異步或做本地緩存。4. RocketMQ國(guó)產(chǎn)電商級(jí)消息中間件部署、控制臺(tái)到高級(jí)特性全解4.1 RocketMQ的核心模型與工作原理RocketMQ的架構(gòu)和Kafka很相似核心概念包括NameServer、Broker、Topic、Consumer Group。它和Kafka最大的不同是引入了NameServer來(lái)實(shí)現(xiàn)路由管理替代了ZooKeeper。NameServer是一個(gè)輕量級(jí)的注冊(cè)中心Broker啟動(dòng)后向所有NameServer注冊(cè)自己的地址信息生產(chǎn)者和消費(fèi)者通過(guò)NameServer獲取Broker地址。RocketMQ的工作流程可以理解成生產(chǎn)者連接NameServer拉取Topic的路由信息把消息發(fā)往對(duì)應(yīng)的BrokerBroker存儲(chǔ)消息并同步到從節(jié)點(diǎn)如果配置了主從消費(fèi)者同樣先從NameServer獲取路由信息再?gòu)腂roker拉取消息。多臺(tái)NameServer之間不互相通信每臺(tái)NameServer都保存全量的路由信息所以部署時(shí)哪怕掛了一臺(tái)NameServer集群依然可用。4.2 Docker Compose部署RocketMQ與控制臺(tái)RocketMQ的部署在三個(gè)隊(duì)列里算是偏繁瑣的因?yàn)樗琋ameServer、Broker、Console三個(gè)組件。手動(dòng)一臺(tái)臺(tái)搭建非常痛苦我強(qiáng)烈建議用Docker Compose一鍵搞定。新建一個(gè)docker-compose.yml文件參考配置如下version: 3.8 services: namesrv: image: apache/rocketmq:5.1.4 container_name: rocketmq-namesrv ports: - 9876:9876 command: sh mqnamesrv broker: image: apache/rocketmq:5.1.4 container_name: rocketmq-broker depends_on: - namesrv ports: - 10909:10909 - 10911:10911 - 10912:10912 environment: - NAMESRV_ADDRnamesrv:9876 volumes: - ./broker.conf:/home/rocketmq/rocketmq-5.1.4/conf/broker.conf command: sh mqbroker -c /home/rocketmq/rocketmq-5.1.4/conf/broker.conf console: image: apacherocketmq/rocketmq-dashboard:latest container_name: rocketmq-console depends_on: - namesrv ports: - 8080:8080 environment: - JAVA_OPTS-Drocketmq.namesrv.addrnamesrv:9876broker.conf文件至少需要配置這幾項(xiàng)否則生產(chǎn)環(huán)境會(huì)出問(wèn)題brokerClusterNameDefaultCluster brokerNamebroker-a brokerId0 deleteWhen04 fileReservedTime48 brokerRoleASYNC_MASTER flushDiskTypeASYNC_FLUSH # 如果broker和代碼不在同一臺(tái)機(jī)器必須配置這個(gè)地址 brokerIP1你的服務(wù)器公網(wǎng)IP或內(nèi)網(wǎng)IP autoCreateTopicEnabletrue很多人部署完RocketMQ后生產(chǎn)消息報(bào)錯(cuò)業(yè)務(wù)測(cè)試怎么都通不過(guò)絕大多數(shù)都是因?yàn)閎rokerIP1沒(méi)有配置導(dǎo)致客戶端連不上broker。這是RocketMQ安裝里最常見(jiàn)的坑之一。執(zhí)行docker-compose up -d后訪問(wèn)http://localhost:8080 就能看到控制臺(tái)可以查看Topic列表、消費(fèi)進(jìn)度、消息軌跡等。4.3 延遲消息、批量消費(fèi)與事務(wù)消息RocketMQ的延遲消息支持比RabbitMQ原生得多。發(fā)送延遲消息不需要搞死信隊(duì)列直接指定delayLevel即可Message msg new Message(order-topic, order.pay.timeout, 訂單超時(shí)關(guān)閉.getBytes()); // delayLevel5 表示延遲1分鐘 msg.setDelayTimeLevel(5); SendResult sendResult producer.send(msg);RocketMQ預(yù)定義了18個(gè)延遲級(jí)別1s、5s、10s、30s、1m、2m、3m、4m、5m、6m、7m、8m、9m、10m、20m、30m、1h、2h分別對(duì)應(yīng)delayLevel的1到18。這個(gè)是固定配置不支持任意時(shí)間。如果項(xiàng)目需要自定義延遲時(shí)間得自己實(shí)現(xiàn)。批量消費(fèi)是RocketMQ在效率方面的一個(gè)優(yōu)勢(shì)。消費(fèi)端可以設(shè)置一次拉取的消息條數(shù)consumer.setConsumeMessageBatchMaxSize(10);配合批量處理能顯著減少網(wǎng)絡(luò)開(kāi)銷和數(shù)據(jù)庫(kù)連接消耗。我在高吞吐業(yè)務(wù)里用過(guò)消費(fèi)TPS提升非常明顯。注意批量消費(fèi)時(shí)是按List傳入你要正確處理List里的每一條消息某一條處理失敗會(huì)影響一批。事務(wù)消息是RocketMQ最引以為傲的特性。它的原理是兩階段提交先發(fā)送半消息Half Message這條消息對(duì)消費(fèi)者不可見(jiàn)然后執(zhí)行本地事務(wù)根據(jù)本地事務(wù)結(jié)果執(zhí)行commit或rollback。如果本地事務(wù)執(zhí)行后出現(xiàn)了超時(shí)RocketMQ會(huì)回調(diào)你注冊(cè)的檢查方法根據(jù)本地事務(wù)的狀態(tài)來(lái)確認(rèn)最終是提交還是回滾。這個(gè)特性能完美解決本地?cái)?shù)據(jù)庫(kù)操作和發(fā)消息不一致的問(wèn)題。比如訂單創(chuàng)建成功后要發(fā)一條優(yōu)惠券消息如果先插數(shù)據(jù)庫(kù)再發(fā)消息可能消息發(fā)送失敗導(dǎo)致用戶收不到券如果先發(fā)消息再操作數(shù)據(jù)庫(kù)可能數(shù)據(jù)庫(kù)操作失敗但消息已經(jīng)發(fā)出去了。用事務(wù)消息可以保證數(shù)據(jù)庫(kù)訂單沒(méi)創(chuàng)建成功消息一定不會(huì)發(fā)出去。4.4 RocketMQ集群模式與生產(chǎn)環(huán)境注意事項(xiàng)RocketMQ的集群模式主要分單主、多主、主從等幾種。單主節(jié)點(diǎn)適合測(cè)試環(huán)境生產(chǎn)環(huán)境至少要做主從。最常用的生產(chǎn)方案是多主多從異步復(fù)制兼顧了吞吐量和數(shù)據(jù)安全。集群部署時(shí)每個(gè)Broker節(jié)點(diǎn)配置不同的brokerName和brokerIdbrokerId為0的代表Master大于0的代表Slave。比如有兩組brokerbroker-a: brokerNamebroker-a, brokerId0broker-a-slave: brokerNamebroker-a, brokerId1broker-b: brokerNamebroker-b, brokerId0broker-b-slave: brokerNamebroker-b, brokerId1生產(chǎn)環(huán)境下還要注意幾個(gè)細(xì)節(jié)autoCreateTopicEnable建議在生產(chǎn)設(shè)為false避免誤創(chuàng)建大量無(wú)意義的TopicfileReservedTime決定消息文件保留多少天超過(guò)會(huì)定期刪除deleteWhen表示凌晨幾點(diǎn)執(zhí)行清理flushDiskType建議生產(chǎn)環(huán)境用SYNC_FLUSH保證消息寫入磁盤成功才返回成功雖然性能有損耗但可靠行更高。5. 三大隊(duì)列的對(duì)比總結(jié)、面試高頻考點(diǎn)與最終選型建議5.1 功能維度橫向?qū)Ρ任野讶齻€(gè)隊(duì)列的核心特性整理成一個(gè)表方便你對(duì)比查看。對(duì)比維度KafkaRabbitMQRocketMQ語(yǔ)言Java/ScalaErlangJava吞吐量極高百萬(wàn)級(jí)/秒中等萬(wàn)級(jí)/秒高十萬(wàn)級(jí)/秒延遲毫秒級(jí)微秒級(jí)毫秒級(jí)消息順序分區(qū)內(nèi)有序單隊(duì)列有序隊(duì)列內(nèi)有序延遲消息不支持原生通過(guò)死信實(shí)現(xiàn)原生支持18個(gè)級(jí)別事務(wù)消息不支持不支持支持死信隊(duì)列需自己實(shí)現(xiàn)原生支持原生支持管理界面無(wú)官方需第三方自帶Web控制臺(tái)官方Dashboard成熟度極高極高高國(guó)內(nèi)電商廣泛使用適用場(chǎng)景日志、大數(shù)據(jù)、流處理業(yè)務(wù)系統(tǒng)、靈活路由電商、金融、業(yè)務(wù)量大5.2 面試高頻考點(diǎn)重復(fù)消費(fèi)、順序消息、消息堆積面試題里三大隊(duì)列經(jīng)常被放在一起對(duì)比但核心考點(diǎn)其實(shí)是幾個(gè)通用的消息中間件問(wèn)題。搞清楚這幾個(gè)問(wèn)題不管用哪個(gè)隊(duì)列都能應(yīng)對(duì)自如。第一個(gè)是消息隊(duì)列重復(fù)消費(fèi)問(wèn)題。這個(gè)我在前面已經(jīng)提過(guò)解決思路核心是消費(fèi)者冪等。面試官更想聽(tīng)的不只是用Redis去重而是你能不能講清楚重復(fù)消費(fèi)產(chǎn)生的根本原因——消息隊(duì)列的At Least Once語(yǔ)義下網(wǎng)絡(luò)分區(qū)、消費(fèi)者宕機(jī)、Ack超時(shí)都可能導(dǎo)致消息重發(fā)。解決思路可以分層數(shù)據(jù)庫(kù)唯一約束、Redis分布式鎖配合業(yè)務(wù)鍵、甚至業(yè)務(wù)邏輯本身就設(shè)計(jì)成天然冪等。第二個(gè)是順序消息怎么保證。Kafka的做法是讓同一業(yè)務(wù)key的消息進(jìn)入同一個(gè)Partition消費(fèi)者單線程消費(fèi)該P(yáng)artition這樣在分區(qū)內(nèi)保持順序。RabbitMQ單隊(duì)列天然有序但引入多個(gè)消費(fèi)者并發(fā)消費(fèi)后順序會(huì)亂需要用一致性哈希把同一業(yè)務(wù)的key路由到同一個(gè)隊(duì)列。RocketMQ提供了普通順序消息和嚴(yán)格順序消息通過(guò)MessageQueueSelector把同一業(yè)務(wù)的key發(fā)送到同一個(gè)隊(duì)列。第三個(gè)是消息堆積如何處理。核心是先保證不丟消息再提高消費(fèi)速度。具體做法臨時(shí)擴(kuò)容消費(fèi)者實(shí)例數(shù)增加單個(gè)消費(fèi)者內(nèi)部的并發(fā)線程或者優(yōu)化消費(fèi)邏輯。如果積壓太嚴(yán)重Kafka可以新建一個(gè)Topic將積壓的消息轉(zhuǎn)發(fā)到新Topic用更多的消費(fèi)者并行處理。RocketMQ和RabbitMQ也可以參考類似思路本質(zhì)是把一條長(zhǎng)隊(duì)列拆成多條短隊(duì)列并行消費(fèi)。還有一個(gè)高頻考點(diǎn)是消息丟失問(wèn)題結(jié)合我前面的內(nèi)容從生產(chǎn)者、broker、消費(fèi)者三個(gè)環(huán)節(jié)分別分析即可。生產(chǎn)者確認(rèn)機(jī)制、broker持久化、消費(fèi)者手動(dòng)Ack每個(gè)環(huán)節(jié)都有對(duì)應(yīng)方案。5.3 基于實(shí)際項(xiàng)目的選型建議聊完對(duì)比和技術(shù)細(xì)節(jié)最后說(shuō)說(shuō)我個(gè)人在實(shí)際項(xiàng)目中的選型心得。如果你在做一個(gè)中小型業(yè)務(wù)系統(tǒng)團(tuán)隊(duì)不大也沒(méi)有復(fù)雜的消息需求RabbitMQ是最穩(wěn)的選擇。它的學(xué)習(xí)曲線平緩、文檔完善、部署簡(jiǎn)單出問(wèn)題也好排查網(wǎng)上答案一搜一大把。如果你所在公司有大數(shù)據(jù)平臺(tái)需要把業(yè)務(wù)數(shù)據(jù)同步到數(shù)倉(cāng)或日志系統(tǒng)或者要做流計(jì)算Kafka幾乎是唯一選項(xiàng)。Kafka的生態(tài)太強(qiáng)了Flink、Spark Streaming這些流處理框架都和它深度集成。如果你在做電商、金融、支付這類高一致性要求的業(yè)務(wù)系統(tǒng)且團(tuán)隊(duì)以Java為主RocketMQ值得認(rèn)真考慮。事務(wù)消息在真實(shí)業(yè)務(wù)里非常實(shí)用能省去大量手工處理本地事務(wù)和消息不一致的代碼。部署方面我的建議是所有消息隊(duì)列都先用Docker或Docker Compose在測(cè)試環(huán)境跑通確認(rèn)無(wú)誤后再考慮生產(chǎn)環(huán)境的物理機(jī)或K8s部署。生產(chǎn)環(huán)境務(wù)必要做集群、監(jiān)控和告警這三個(gè)隊(duì)列都有對(duì)應(yīng)的監(jiān)控手段RabbitMQ看管理臺(tái)的隊(duì)列堆積和消費(fèi)速率Kafka看Consumer Group的Lag和broker的存活狀態(tài)RocketMQ看控制臺(tái)的消費(fèi)進(jìn)度和消息軌跡。6. 寫在最后我的實(shí)戰(zhàn)心得與排障技巧從我這些年折騰三個(gè)隊(duì)列的經(jīng)驗(yàn)來(lái)看消息隊(duì)列本身不難難的是把消息隊(duì)列用得穩(wěn)。很多人剛開(kāi)始用的時(shí)候只關(guān)心怎么把消息發(fā)出去、怎么收下來(lái)等上了生產(chǎn)環(huán)境遇到消息丟失、重復(fù)消費(fèi)、順序錯(cuò)亂才開(kāi)始頭疼。所以我最后針對(duì)三個(gè)隊(duì)列各分享一個(gè)容易被人忽略的實(shí)戰(zhàn)技巧。RabbitMQ這個(gè)技巧是關(guān)于連接管理的。生產(chǎn)環(huán)境一定要用連接池管理Connection不要每次發(fā)消息都新建連接。Connection的創(chuàng)建是重操作頻繁創(chuàng)建會(huì)打滿文件描述符最終導(dǎo)致連接被拒絕。Channel也是類似同一個(gè)線程可以復(fù)用同一個(gè)Channel。Kafka的實(shí)戰(zhàn)技巧是關(guān)于Consumer的。消費(fèi)者啟動(dòng)后要注意設(shè)置一個(gè)合理的max.poll.interval.ms默認(rèn)值是5分鐘。如果你的單條消息處理時(shí)間超過(guò)5分鐘消費(fèi)者會(huì)被判定為假死觸發(fā)rebalance消息會(huì)被其他消費(fèi)者重新消費(fèi)極容易引發(fā)重復(fù)消費(fèi)。遇到過(guò)好幾次莫名重復(fù)消費(fèi)的線上事故排查到最后都是這個(gè)參數(shù)的問(wèn)題。RocketMQ的實(shí)戰(zhàn)技巧是關(guān)于消息軌跡的。如果線上消息莫名其妙就沒(méi)了先在控制臺(tái)的消息軌跡里查一下消息的完整生命周期看是生產(chǎn)端沒(méi)發(fā)出去、broker丟失還是消費(fèi)端拉到了但處理失敗。RocketMQ的軌跡功能非常完善定位問(wèn)題效率很高。不要一上來(lái)就去翻服務(wù)日志那樣太慢。最后再講一個(gè)所有消息隊(duì)列都適用的排查思路消息隊(duì)列的問(wèn)題本質(zhì)上是鏈路問(wèn)題。排查時(shí)一定不要只看單個(gè)節(jié)點(diǎn)要從生產(chǎn)者、broker、消費(fèi)者三段同時(shí)看哪一段有瓶頸數(shù)據(jù)會(huì)說(shuō)話的。把消息隊(duì)列的指標(biāo)監(jiān)控做起來(lái)比什么大牛排查技巧都管用。 三大消息隊(duì)列Kafka、RabbitMQ、RocketMQ選型對(duì)比與實(shí)戰(zhàn)指南
先說(shuō)個(gè)背景。做后端這幾年消息隊(duì)列基本是繞不開(kāi)的中間件。不管是電商的訂單流轉(zhuǎn)、日志收集、異步通知還是高并發(fā)下的削峰填谷你總會(huì)遇到一個(gè)場(chǎng)景需要引入消息隊(duì)列。市面上一堆MQ但真正經(jīng)常被提起、被面試官追問(wèn)、被生產(chǎn)環(huán)境大規(guī)模使用的無(wú)非就是Kafka、RabbitMQ和RocketMQ這三個(gè)。我這些年三個(gè)隊(duì)列都用過(guò)踩過(guò)不少坑也總結(jié)出一些實(shí)戰(zhàn)經(jīng)驗(yàn)這篇就一次性把它們從選型、部署、代碼接入到問(wèn)題排查完整串一遍。這篇內(nèi)容不會(huì)照著官方文檔念而是從實(shí)際使用者的視角出發(fā)講清楚每個(gè)隊(duì)列的核心原理、安裝關(guān)鍵點(diǎn)和代碼落地的真實(shí)細(xì)節(jié)。不管你是剛接觸消息隊(duì)列的新人還是已經(jīng)用了一段時(shí)間、想系統(tǒng)補(bǔ)一遍的開(kāi)發(fā)者都能從中找到可直接參考的東西。文章比較長(zhǎng)建議先收藏等真正用的時(shí)候再回來(lái)翻。1. 提筆之前消息隊(duì)列到底解決什么問(wèn)題三大隊(duì)列該怎么選1.1 消息隊(duì)列的核心價(jià)值不是隊(duì)列而是解耦很多人理解消息隊(duì)列第一反應(yīng)就是先進(jìn)先出的數(shù)據(jù)結(jié)構(gòu)。這個(gè)理解沒(méi)錯(cuò)但在實(shí)際工程里消息隊(duì)列的核心價(jià)值根本不是隊(duì)列本身而是它帶來(lái)的三個(gè)能力異步、削峰、解耦。我再拆細(xì)一點(diǎn)。異步用戶下單后系統(tǒng)需要發(fā)短信、發(fā)優(yōu)惠券、更新積分。如果這些步驟全部串行執(zhí)行一次請(qǐng)求可能要等幾百毫秒甚至更久。把非核心操作丟進(jìn)隊(duì)列主鏈路只處理扣庫(kù)存、生成訂單響應(yīng)時(shí)間直接從秒級(jí)降到毫秒級(jí)。削峰秒殺場(chǎng)景下瞬間涌入十萬(wàn)請(qǐng)求數(shù)據(jù)庫(kù)根本扛不住。讓請(qǐng)求先進(jìn)隊(duì)列后端按自身處理能力慢慢消費(fèi)系統(tǒng)就不會(huì)被流量打垮。解耦訂單服務(wù)不需要知道庫(kù)存服務(wù)、物流服務(wù)的具體實(shí)現(xiàn)只要往隊(duì)列里發(fā)一條訂單創(chuàng)建成功的消息下游服務(wù)各自訂閱自己關(guān)心的內(nèi)容互不干擾。這也是為什么現(xiàn)在的面試題里消息隊(duì)列幾乎是必問(wèn)項(xiàng)。面試官想知道的不只是你會(huì)不會(huì)調(diào)API而是你有沒(méi)有真正理解它解決什么問(wèn)題、在什么場(chǎng)景下選哪個(gè)。1.2 三個(gè)隊(duì)列的定位差異與選型邏輯Kafka、RabbitMQ、RocketMQ雖然都叫消息隊(duì)列但出生背景完全不同這也決定了它們的主戰(zhàn)場(chǎng)不同。Kafka出生在LinkedIn最初是為了處理海量日志流而設(shè)計(jì)的。它的架構(gòu)天然適合大規(guī)模數(shù)據(jù)管道、日志采集、流式計(jì)算這類場(chǎng)景吞吐量極高但功能相對(duì)寡淡——延時(shí)消息、死信隊(duì)列這些高級(jí)特性原生支持都不太好得靠外部或自己實(shí)現(xiàn)。RabbitMQ是Erlang寫的老牌消息中間件功能非常豐富路由模式靈活文檔生態(tài)完善適合企業(yè)內(nèi)部業(yè)務(wù)系統(tǒng)之間做消息通信。RocketMQ是阿里開(kāi)源的消息中間件人家是從電商業(yè)務(wù)里長(zhǎng)出來(lái)的所以天然補(bǔ)齊了阿里巴巴在交易鏈路上的各種需求事務(wù)消息、延時(shí)消息、消息重試、死信隊(duì)列甚至消息軌跡追蹤都有。簡(jiǎn)單總結(jié)選型邏輯你們做的是大數(shù)據(jù)管道、日志采集、流計(jì)算那一路選Kafka業(yè)務(wù)系統(tǒng)內(nèi)部需要靈活的路由、優(yōu)先級(jí)、RPC調(diào)用通知選RabbitMQ業(yè)務(wù)量很大并且深度使用Java需要事務(wù)消息和可靠投遞選RocketMQ。這不是絕對(duì)的但九成場(chǎng)景這么選都不會(huì)錯(cuò)。2. RabbitMQ功能最全的業(yè)務(wù)消息中間件從安裝到生產(chǎn)落地2.1 安裝部署Windows、Linux和Docker三種方式對(duì)比RabbitMQ在Windows上的安裝以前是不少人入門的第一道坎。它依賴Erlang環(huán)境需要先下載Erlang的Windows安裝包再下載RabbitMQ Server的Windows安裝包而且版本要對(duì)應(yīng)。網(wǎng)上關(guān)于RabbitMQ下載和安裝的教程很多但我見(jiàn)過(guò)太多人卡在Erlang版本和RabbitMQ版本不匹配這個(gè)問(wèn)題上。官網(wǎng)上每個(gè)RabbitMQ版本都會(huì)標(biāo)注對(duì)應(yīng)的Erlang版本區(qū)間安裝前務(wù)必核對(duì)。Windows下安裝完還需要手動(dòng)激活管理插件命令行執(zhí)行rabbitmq-plugins enable rabbitmq_management然后重啟RabbitMQ服務(wù)瀏覽器訪問(wèn)http://localhost:15672 用默認(rèn)賬號(hào)guest/guest登錄。這里有個(gè)坑默認(rèn)的guest賬號(hào)只能在localhost訪問(wèn)如果你用IP地址訪問(wèn)管理臺(tái)會(huì)被拒絕。要么用localhost訪問(wèn)要么新建一個(gè)用戶并賦予權(quán)限。Linux下安裝稍微順暢一些因?yàn)楣俜教峁┝薬pt和yum源。以Ubuntu為例sudo apt-get install -y rabbitmq-server sudo systemctl enable rabbitmq-server sudo systemctl start rabbitmq-server sudo rabbitmq-plugins enable rabbitmq_management但最推薦的方式還是Docker尤其適合本地開(kāi)發(fā)和測(cè)試環(huán)境。一條命令搞定還不用操心底層依賴docker run -d --name rabbitmq \ -p 5672:5672 -p 15672:15672 \ -e RABBITMQ_DEFAULT_USERadmin \ -e RABBITMQ_DEFAULT_PASSadmin123 \ rabbitmq:3.12-management這里要注意鏡像分為兩個(gè)版本rabbitmq版本號(hào)不帶management和rabbitmq版本號(hào)-management帶管理臺(tái)插件。本地調(diào)試建議直接用management版本省得手動(dòng)裝插件。2.2 核心概念與Java客戶端接入RabbitMQ的核心模型是生產(chǎn)者Producer把消息發(fā)給交換機(jī)Exchange交換機(jī)根據(jù)路由規(guī)則把消息投遞到綁定的隊(duì)列Queue消費(fèi)者Consumer從隊(duì)列拉取消息處理。很多人第一反應(yīng)是為什么不直接發(fā)到隊(duì)列原因在于交換機(jī)提供了靈活的路由能力。交換機(jī)有四種類型Direct點(diǎn)對(duì)點(diǎn)精確匹配、Fanout廣播給所有綁定的隊(duì)列、Topic通配符匹配路由鍵、Headers基于消息頭的匹配用得少。實(shí)際工程中Topic交換機(jī)用得最多因?yàn)樗钠ヅ湟?guī)則靈活。比如訂單系統(tǒng)發(fā)一條消息路由鍵是order.created物流服務(wù)綁定order.*積分服務(wù)綁定order.created兩個(gè)服務(wù)都能收到消息非常方便。Java客戶端接入的代碼核心就三塊。第一塊是創(chuàng)建連接工廠ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); factory.setUsername(admin); factory.setPassword(admin123); Connection connection factory.newConnection(); Channel channel connection.createChannel();第二塊是聲明隊(duì)列和交換機(jī)并做綁定channel.exchangeDeclare(order.exchange, topic, true); channel.queueDeclare(order.queue, true, false, false, null); channel.queueBind(order.queue, order.exchange, order.#);第三塊是發(fā)送消息和接收消息。發(fā)送String message 訂單創(chuàng)建成功; channel.basicPublish(order.exchange, order.created, null, message.getBytes());接收channel.basicConsume(order.queue, false, (consumerTag, delivery) - { String msg new String(delivery.getBody(), UTF-8); System.out.println(收到消息: msg); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); }, consumerTag - {});注意basicConsume的第二個(gè)參數(shù)是autoAck工作中我建議一律顯式設(shè)為false手動(dòng)調(diào)用basicAck確認(rèn)消息。原因很簡(jiǎn)單如果設(shè)置了自動(dòng)確認(rèn)消費(fèi)者收到消息但還沒(méi)來(lái)得及處理就掛了消息就丟了手動(dòng)確認(rèn)可以確保消息被完整處理后才會(huì)從隊(duì)列移除。2.3 消息可靠性生產(chǎn)端確認(rèn)與消費(fèi)端手動(dòng)Ack消息隊(duì)列最讓人頭疼的問(wèn)題就是消息丟失。RabbitMQ的消息丟失可能發(fā)生在三個(gè)環(huán)節(jié)生產(chǎn)者發(fā)送時(shí)、隊(duì)列存儲(chǔ)時(shí)、消費(fèi)者消費(fèi)時(shí)。生產(chǎn)端要開(kāi)啟生產(chǎn)確認(rèn)模式。在channel上調(diào)用confirmSelect()方法發(fā)布消息后等待broker的確認(rèn)回調(diào)不確認(rèn)就重發(fā)。這是第一層保障。隊(duì)列持久化第二層保障聲明隊(duì)列時(shí)把durable參數(shù)設(shè)為true這樣RabbitMQ重啟后隊(duì)列和消息還在。消費(fèi)端手動(dòng)Ack第三層保障前面已經(jīng)提到過(guò)。這三層都做好了消息才能做到不丟失。很多人配置完不管不顧直到某天凌晨收到告警郵件才發(fā)現(xiàn)消息丟了一查結(jié)果就是消費(fèi)者沒(méi)做手動(dòng)確認(rèn)。這類問(wèn)題我在生產(chǎn)環(huán)境見(jiàn)過(guò)太多次。2.4 高頻問(wèn)題延時(shí)隊(duì)列、死信隊(duì)列與重復(fù)消費(fèi)RabbitMQ原生的延時(shí)隊(duì)列是通過(guò)死信交換機(jī)DLX機(jī)制實(shí)現(xiàn)的。思路是給隊(duì)列設(shè)置一個(gè)TTL消息過(guò)期后會(huì)被轉(zhuǎn)投到綁定的死信交換機(jī)再由死信交換機(jī)路由到真正的處理隊(duì)列。這個(gè)方案雖然繞但生產(chǎn)環(huán)境很常見(jiàn)。比如電商下單后30分鐘未支付自動(dòng)關(guān)閉訂單就是用延時(shí)隊(duì)列實(shí)現(xiàn)的。所以RabbitMQ面試題里死信隊(duì)列是絕對(duì)的高頻考點(diǎn)。實(shí)現(xiàn)要點(diǎn)是創(chuàng)建兩個(gè)隊(duì)列和一個(gè)交換機(jī)一個(gè)普通隊(duì)列帶DLX配置和TTL一個(gè)實(shí)際處理隊(duì)列綁定死信交換機(jī)。重復(fù)消費(fèi)問(wèn)題則是所有消息隊(duì)列的通病。根本原因在于分布式網(wǎng)絡(luò)環(huán)境下消息可能因?yàn)榫W(wǎng)絡(luò)抖動(dòng)導(dǎo)致消費(fèi)者已經(jīng)處理完但Ack沒(méi)送達(dá)brokerbroker會(huì)重發(fā)消息。解決思路也比較統(tǒng)一消費(fèi)者側(cè)做冪等處理。常見(jiàn)方案有用redis setnx做冪等鍵、數(shù)據(jù)庫(kù)唯一約束、業(yè)務(wù)表里加一個(gè)處理狀態(tài)字段。我在項(xiàng)目里用的最多的是業(yè)務(wù)鍵Redis的方式消息到達(dá)先檢查Redis里是否已存在該消息ID存在則直接返回不存在則處理并寫入Redis。3. Kafka高吞吐的日志與流數(shù)據(jù)處理利器原理和集群安裝一次說(shuō)清3.1 Kafka的架構(gòu)與工作原理先弄懂Kafka的基本模型BrokerKafka服務(wù)器節(jié)點(diǎn)、Topic消息分類、PartitionTopic的分區(qū)、Producer生產(chǎn)者、Consumer消費(fèi)者和Consumer Group消費(fèi)者組。Kafka最大的特點(diǎn)是分區(qū)并行這也是它吞吐量遠(yuǎn)超其他MQ的關(guān)鍵。一個(gè)Topic可以分成多個(gè)Partition每個(gè)Partition是獨(dú)立的日志文件讀寫互不干擾。消費(fèi)者組里的每個(gè)消費(fèi)者負(fù)責(zé)部分分區(qū)實(shí)現(xiàn)水平擴(kuò)展。Kafka的消息在Partition內(nèi)是有序的不同Partition之間無(wú)順序保證。Kafka的存儲(chǔ)機(jī)制也很特別。消息寫入后不是立即刪除而是保存在磁盤里基于順序?qū)懞晚?yè)緩存技術(shù)保證高吞吐。消息的消費(fèi)進(jìn)度由消費(fèi)者自己管理——記錄當(dāng)前消費(fèi)到哪個(gè)Offset。這也解釋了為什么Kafka的消費(fèi)者可以做到隨時(shí)回溯你可以把Offset重置到昨天重新消費(fèi)一遍。3.2 單機(jī)與集群安裝Kafka本身依賴ZooKeeper新版雖然引入了KRaft模式但大多數(shù)生產(chǎn)環(huán)境還在用ZooKeeper安裝Kafka基本離不開(kāi)ZooKeeper。這也是Kafka安裝鏈路比其他MQ繁重的原因。單機(jī)安裝比較簡(jiǎn)單這里是Linux環(huán)境操作# 下載解壓kafka wget https://downloads.apache.org/kafka/3.4.0/kafka_2.13-3.4.0.tgz tar -zxvf kafka_2.13-3.4.0.tgz cd kafka_2.13-3.4.0 # 啟動(dòng)ZooKeeper bin/zookeeper-server-start.sh config/zookeeper.properties # 啟動(dòng)Kafka bin/kafka-server-start.sh config/server.properties 集群安裝本質(zhì)上就是多節(jié)點(diǎn)配置。每個(gè)Broker要修改config/server.properties里的幾個(gè)關(guān)鍵參數(shù)broker.id集群內(nèi)唯一、listeners監(jiān)聽(tīng)地址、log.dirs日志目錄、zookeeper.connect指向ZooKeeper集群。三節(jié)點(diǎn)的集群配置大致是這樣node1: broker.id0listenersPLAINTEXT://192.168.1.10:9092node2: broker.id1listenersPLAINTEXT://192.168.1.11:9092node3: broker.id2listenersPLAINTEXT://192.168.1.12:9092ZooKeeper地址都寫成zookeeper集群的地址列表比如192.168.1.1:2181,192.168.1.2:2181,192.168.1.3:2181。如果是離線環(huán)境Kafka集群離線安裝也不復(fù)雜——就是把已經(jīng)下載好的tar包和依賴包通過(guò)內(nèi)網(wǎng)傳輸?shù)侥繕?biāo)機(jī)器安裝步驟和在線一致。關(guān)鍵在于三臺(tái)機(jī)器的時(shí)鐘要同步建議配置NTP。集群配置完成后用kafka-topics.sh創(chuàng)建topic時(shí)指定replication-factor數(shù)據(jù)才會(huì)在多個(gè)broker間復(fù)制。3.3 生產(chǎn)與消費(fèi)命令實(shí)操Kafka的命令行是排查問(wèn)題的利器熟練掌握能省很多時(shí)間。創(chuàng)建主題bin/kafka-topics.sh --create \ --bootstrap-server localhost:9092 \ --replication-factor 1 --partitions 3 \ --topic order-topic生產(chǎn)消息測(cè)試bin/kafka-console-producer.sh --broker-list localhost:9092 --topic order-topic消費(fèi)消息bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \ --topic order-topic --from-beginning這條消費(fèi)命令有一個(gè)參數(shù)經(jīng)常被忽略--from-beginning。它的含義是從最早的Offset開(kāi)始消費(fèi)。如果你想讓消費(fèi)者從指定時(shí)間點(diǎn)開(kāi)始消費(fèi)可以加--offset 指定Offset位置或者用--partition加--offset組合。熱搜詞里提到的kafka消費(fèi)命令指定消費(fèi)時(shí)間實(shí)際做法是用kafka-consumer-groups.sh重置偏移量bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \ --group my-group --topic order-topic \ --reset-offsets --to-datetime 2024-01-01T00:00:00.000 --execute這條命令很實(shí)用當(dāng)年排查線上數(shù)據(jù)異常時(shí)就是用這個(gè)把消費(fèi)者組的Offset重置到出問(wèn)題之前的某個(gè)時(shí)間點(diǎn)讓消費(fèi)者重新消費(fèi)那段歷史消息。3.4 Kafka可視化工具與消息延遲排查Kafka是出了名的難以可視化的中間件不像RabbitMQ自帶一個(gè)漂亮的管理臺(tái)。很多剛接觸Kafka的人都會(huì)搜kafka可視化工具、kafka連接工具、kafka圖形界面。這里我推薦兩個(gè)Kafka Tool現(xiàn)在叫Offset Explorer和Kafka UI。前者是桌面客戶端圖形化查看Topic、Partition、Consumer Group的Offset情況適合日常排查后者是Web界面支持集群管理、消息查看和動(dòng)態(tài)創(chuàng)建Topic適合團(tuán)隊(duì)共用。關(guān)于Kafka消息延遲高的問(wèn)題幾乎每個(gè)團(tuán)隊(duì)都會(huì)遇到。我排查這類問(wèn)題的經(jīng)驗(yàn)是按鏈路逐個(gè)檢查先看生產(chǎn)者端發(fā)送耗時(shí)是不是發(fā)送時(shí)同步等待過(guò)久如果批量發(fā)送參數(shù)設(shè)置不合理會(huì)影響性能。再看消費(fèi)者端處理速度是不是單條消息處理邏輯太慢或者消費(fèi)者線程數(shù)太低。重點(diǎn)看兩個(gè)指標(biāo)Consumer Group的Lag積壓量和每個(gè)分區(qū)的處理耗時(shí)。如果Lag持續(xù)增長(zhǎng)說(shuō)明消費(fèi)速度趕不上生產(chǎn)速度。處理人最多的方案是加消費(fèi)者實(shí)例增加單個(gè)消費(fèi)者的線程數(shù)或者優(yōu)化消費(fèi)邏輯的DB訪問(wèn)比如把單條插入改成批量插入。還有一點(diǎn)很容易被忽略如果某些消息處理邏輯里有外部API調(diào)用而外部服務(wù)響應(yīng)慢會(huì)直接拖垮整個(gè)消費(fèi)者的消費(fèi)速度。這種情況建議把外部調(diào)用改成異步或做本地緩存。4. RocketMQ國(guó)產(chǎn)電商級(jí)消息中間件部署、控制臺(tái)到高級(jí)特性全解4.1 RocketMQ的核心模型與工作原理RocketMQ的架構(gòu)和Kafka很相似核心概念包括NameServer、Broker、Topic、Consumer Group。它和Kafka最大的不同是引入了NameServer來(lái)實(shí)現(xiàn)路由管理替代了ZooKeeper。NameServer是一個(gè)輕量級(jí)的注冊(cè)中心Broker啟動(dòng)后向所有NameServer注冊(cè)自己的地址信息生產(chǎn)者和消費(fèi)者通過(guò)NameServer獲取Broker地址。RocketMQ的工作流程可以理解成生產(chǎn)者連接NameServer拉取Topic的路由信息把消息發(fā)往對(duì)應(yīng)的BrokerBroker存儲(chǔ)消息并同步到從節(jié)點(diǎn)如果配置了主從消費(fèi)者同樣先從NameServer獲取路由信息再?gòu)腂roker拉取消息。多臺(tái)NameServer之間不互相通信每臺(tái)NameServer都保存全量的路由信息所以部署時(shí)哪怕掛了一臺(tái)NameServer集群依然可用。4.2 Docker Compose部署RocketMQ與控制臺(tái)RocketMQ的部署在三個(gè)隊(duì)列里算是偏繁瑣的因?yàn)樗琋ameServer、Broker、Console三個(gè)組件。手動(dòng)一臺(tái)臺(tái)搭建非常痛苦我強(qiáng)烈建議用Docker Compose一鍵搞定。新建一個(gè)docker-compose.yml文件參考配置如下version: 3.8 services: namesrv: image: apache/rocketmq:5.1.4 container_name: rocketmq-namesrv ports: - 9876:9876 command: sh mqnamesrv broker: image: apache/rocketmq:5.1.4 container_name: rocketmq-broker depends_on: - namesrv ports: - 10909:10909 - 10911:10911 - 10912:10912 environment: - NAMESRV_ADDRnamesrv:9876 volumes: - ./broker.conf:/home/rocketmq/rocketmq-5.1.4/conf/broker.conf command: sh mqbroker -c /home/rocketmq/rocketmq-5.1.4/conf/broker.conf console: image: apacherocketmq/rocketmq-dashboard:latest container_name: rocketmq-console depends_on: - namesrv ports: - 8080:8080 environment: - JAVA_OPTS-Drocketmq.namesrv.addrnamesrv:9876broker.conf文件至少需要配置這幾項(xiàng)否則生產(chǎn)環(huán)境會(huì)出問(wèn)題brokerClusterNameDefaultCluster brokerNamebroker-a brokerId0 deleteWhen04 fileReservedTime48 brokerRoleASYNC_MASTER flushDiskTypeASYNC_FLUSH # 如果broker和代碼不在同一臺(tái)機(jī)器必須配置這個(gè)地址 brokerIP1你的服務(wù)器公網(wǎng)IP或內(nèi)網(wǎng)IP autoCreateTopicEnabletrue很多人部署完RocketMQ后生產(chǎn)消息報(bào)錯(cuò)業(yè)務(wù)測(cè)試怎么都通不過(guò)絕大多數(shù)都是因?yàn)閎rokerIP1沒(méi)有配置導(dǎo)致客戶端連不上broker。這是RocketMQ安裝里最常見(jiàn)的坑之一。執(zhí)行docker-compose up -d后訪問(wèn)http://localhost:8080 就能看到控制臺(tái)可以查看Topic列表、消費(fèi)進(jìn)度、消息軌跡等。4.3 延遲消息、批量消費(fèi)與事務(wù)消息RocketMQ的延遲消息支持比RabbitMQ原生得多。發(fā)送延遲消息不需要搞死信隊(duì)列直接指定delayLevel即可Message msg new Message(order-topic, order.pay.timeout, 訂單超時(shí)關(guān)閉.getBytes()); // delayLevel5 表示延遲1分鐘 msg.setDelayTimeLevel(5); SendResult sendResult producer.send(msg);RocketMQ預(yù)定義了18個(gè)延遲級(jí)別1s、5s、10s、30s、1m、2m、3m、4m、5m、6m、7m、8m、9m、10m、20m、30m、1h、2h分別對(duì)應(yīng)delayLevel的1到18。這個(gè)是固定配置不支持任意時(shí)間。如果項(xiàng)目需要自定義延遲時(shí)間得自己實(shí)現(xiàn)。批量消費(fèi)是RocketMQ在效率方面的一個(gè)優(yōu)勢(shì)。消費(fèi)端可以設(shè)置一次拉取的消息條數(shù)consumer.setConsumeMessageBatchMaxSize(10);配合批量處理能顯著減少網(wǎng)絡(luò)開(kāi)銷和數(shù)據(jù)庫(kù)連接消耗。我在高吞吐業(yè)務(wù)里用過(guò)消費(fèi)TPS提升非常明顯。注意批量消費(fèi)時(shí)是按List傳入你要正確處理List里的每一條消息某一條處理失敗會(huì)影響一批。事務(wù)消息是RocketMQ最引以為傲的特性。它的原理是兩階段提交先發(fā)送半消息Half Message這條消息對(duì)消費(fèi)者不可見(jiàn)然后執(zhí)行本地事務(wù)根據(jù)本地事務(wù)結(jié)果執(zhí)行commit或rollback。如果本地事務(wù)執(zhí)行后出現(xiàn)了超時(shí)RocketMQ會(huì)回調(diào)你注冊(cè)的檢查方法根據(jù)本地事務(wù)的狀態(tài)來(lái)確認(rèn)最終是提交還是回滾。這個(gè)特性能完美解決本地?cái)?shù)據(jù)庫(kù)操作和發(fā)消息不一致的問(wèn)題。比如訂單創(chuàng)建成功后要發(fā)一條優(yōu)惠券消息如果先插數(shù)據(jù)庫(kù)再發(fā)消息可能消息發(fā)送失敗導(dǎo)致用戶收不到券如果先發(fā)消息再操作數(shù)據(jù)庫(kù)可能數(shù)據(jù)庫(kù)操作失敗但消息已經(jīng)發(fā)出去了。用事務(wù)消息可以保證數(shù)據(jù)庫(kù)訂單沒(méi)創(chuàng)建成功消息一定不會(huì)發(fā)出去。4.4 RocketMQ集群模式與生產(chǎn)環(huán)境注意事項(xiàng)RocketMQ的集群模式主要分單主、多主、主從等幾種。單主節(jié)點(diǎn)適合測(cè)試環(huán)境生產(chǎn)環(huán)境至少要做主從。最常用的生產(chǎn)方案是多主多從異步復(fù)制兼顧了吞吐量和數(shù)據(jù)安全。集群部署時(shí)每個(gè)Broker節(jié)點(diǎn)配置不同的brokerName和brokerIdbrokerId為0的代表Master大于0的代表Slave。比如有兩組brokerbroker-a: brokerNamebroker-a, brokerId0broker-a-slave: brokerNamebroker-a, brokerId1broker-b: brokerNamebroker-b, brokerId0broker-b-slave: brokerNamebroker-b, brokerId1生產(chǎn)環(huán)境下還要注意幾個(gè)細(xì)節(jié)autoCreateTopicEnable建議在生產(chǎn)設(shè)為false避免誤創(chuàng)建大量無(wú)意義的TopicfileReservedTime決定消息文件保留多少天超過(guò)會(huì)定期刪除deleteWhen表示凌晨幾點(diǎn)執(zhí)行清理flushDiskType建議生產(chǎn)環(huán)境用SYNC_FLUSH保證消息寫入磁盤成功才返回成功雖然性能有損耗但可靠行更高。5. 三大隊(duì)列的對(duì)比總結(jié)、面試高頻考點(diǎn)與最終選型建議5.1 功能維度橫向?qū)Ρ任野讶齻€(gè)隊(duì)列的核心特性整理成一個(gè)表方便你對(duì)比查看。對(duì)比維度KafkaRabbitMQRocketMQ語(yǔ)言Java/ScalaErlangJava吞吐量極高百萬(wàn)級(jí)/秒中等萬(wàn)級(jí)/秒高十萬(wàn)級(jí)/秒延遲毫秒級(jí)微秒級(jí)毫秒級(jí)消息順序分區(qū)內(nèi)有序單隊(duì)列有序隊(duì)列內(nèi)有序延遲消息不支持原生通過(guò)死信實(shí)現(xiàn)原生支持18個(gè)級(jí)別事務(wù)消息不支持不支持支持死信隊(duì)列需自己實(shí)現(xiàn)原生支持原生支持管理界面無(wú)官方需第三方自帶Web控制臺(tái)官方Dashboard成熟度極高極高高國(guó)內(nèi)電商廣泛使用適用場(chǎng)景日志、大數(shù)據(jù)、流處理業(yè)務(wù)系統(tǒng)、靈活路由電商、金融、業(yè)務(wù)量大5.2 面試高頻考點(diǎn)重復(fù)消費(fèi)、順序消息、消息堆積面試題里三大隊(duì)列經(jīng)常被放在一起對(duì)比但核心考點(diǎn)其實(shí)是幾個(gè)通用的消息中間件問(wèn)題。搞清楚這幾個(gè)問(wèn)題不管用哪個(gè)隊(duì)列都能應(yīng)對(duì)自如。第一個(gè)是消息隊(duì)列重復(fù)消費(fèi)問(wèn)題。這個(gè)我在前面已經(jīng)提過(guò)解決思路核心是消費(fèi)者冪等。面試官更想聽(tīng)的不只是用Redis去重而是你能不能講清楚重復(fù)消費(fèi)產(chǎn)生的根本原因——消息隊(duì)列的At Least Once語(yǔ)義下網(wǎng)絡(luò)分區(qū)、消費(fèi)者宕機(jī)、Ack超時(shí)都可能導(dǎo)致消息重發(fā)。解決思路可以分層數(shù)據(jù)庫(kù)唯一約束、Redis分布式鎖配合業(yè)務(wù)鍵、甚至業(yè)務(wù)邏輯本身就設(shè)計(jì)成天然冪等。第二個(gè)是順序消息怎么保證。Kafka的做法是讓同一業(yè)務(wù)key的消息進(jìn)入同一個(gè)Partition消費(fèi)者單線程消費(fèi)該P(yáng)artition這樣在分區(qū)內(nèi)保持順序。RabbitMQ單隊(duì)列天然有序但引入多個(gè)消費(fèi)者并發(fā)消費(fèi)后順序會(huì)亂需要用一致性哈希把同一業(yè)務(wù)的key路由到同一個(gè)隊(duì)列。RocketMQ提供了普通順序消息和嚴(yán)格順序消息通過(guò)MessageQueueSelector把同一業(yè)務(wù)的key發(fā)送到同一個(gè)隊(duì)列。第三個(gè)是消息堆積如何處理。核心是先保證不丟消息再提高消費(fèi)速度。具體做法臨時(shí)擴(kuò)容消費(fèi)者實(shí)例數(shù)增加單個(gè)消費(fèi)者內(nèi)部的并發(fā)線程或者優(yōu)化消費(fèi)邏輯。如果積壓太嚴(yán)重Kafka可以新建一個(gè)Topic將積壓的消息轉(zhuǎn)發(fā)到新Topic用更多的消費(fèi)者并行處理。RocketMQ和RabbitMQ也可以參考類似思路本質(zhì)是把一條長(zhǎng)隊(duì)列拆成多條短隊(duì)列并行消費(fèi)。還有一個(gè)高頻考點(diǎn)是消息丟失問(wèn)題結(jié)合我前面的內(nèi)容從生產(chǎn)者、broker、消費(fèi)者三個(gè)環(huán)節(jié)分別分析即可。生產(chǎn)者確認(rèn)機(jī)制、broker持久化、消費(fèi)者手動(dòng)Ack每個(gè)環(huán)節(jié)都有對(duì)應(yīng)方案。5.3 基于實(shí)際項(xiàng)目的選型建議聊完對(duì)比和技術(shù)細(xì)節(jié)最后說(shuō)說(shuō)我個(gè)人在實(shí)際項(xiàng)目中的選型心得。如果你在做一個(gè)中小型業(yè)務(wù)系統(tǒng)團(tuán)隊(duì)不大也沒(méi)有復(fù)雜的消息需求RabbitMQ是最穩(wěn)的選擇。它的學(xué)習(xí)曲線平緩、文檔完善、部署簡(jiǎn)單出問(wèn)題也好排查網(wǎng)上答案一搜一大把。如果你所在公司有大數(shù)據(jù)平臺(tái)需要把業(yè)務(wù)數(shù)據(jù)同步到數(shù)倉(cāng)或日志系統(tǒng)或者要做流計(jì)算Kafka幾乎是唯一選項(xiàng)。Kafka的生態(tài)太強(qiáng)了Flink、Spark Streaming這些流處理框架都和它深度集成。如果你在做電商、金融、支付這類高一致性要求的業(yè)務(wù)系統(tǒng)且團(tuán)隊(duì)以Java為主RocketMQ值得認(rèn)真考慮。事務(wù)消息在真實(shí)業(yè)務(wù)里非常實(shí)用能省去大量手工處理本地事務(wù)和消息不一致的代碼。部署方面我的建議是所有消息隊(duì)列都先用Docker或Docker Compose在測(cè)試環(huán)境跑通確認(rèn)無(wú)誤后再考慮生產(chǎn)環(huán)境的物理機(jī)或K8s部署。生產(chǎn)環(huán)境務(wù)必要做集群、監(jiān)控和告警這三個(gè)隊(duì)列都有對(duì)應(yīng)的監(jiān)控手段RabbitMQ看管理臺(tái)的隊(duì)列堆積和消費(fèi)速率Kafka看Consumer Group的Lag和broker的存活狀態(tài)RocketMQ看控制臺(tái)的消費(fèi)進(jìn)度和消息軌跡。6. 寫在最后我的實(shí)戰(zhàn)心得與排障技巧從我這些年折騰三個(gè)隊(duì)列的經(jīng)驗(yàn)來(lái)看消息隊(duì)列本身不難難的是把消息隊(duì)列用得穩(wěn)。很多人剛開(kāi)始用的時(shí)候只關(guān)心怎么把消息發(fā)出去、怎么收下來(lái)等上了生產(chǎn)環(huán)境遇到消息丟失、重復(fù)消費(fèi)、順序錯(cuò)亂才開(kāi)始頭疼。所以我最后針對(duì)三個(gè)隊(duì)列各分享一個(gè)容易被人忽略的實(shí)戰(zhàn)技巧。RabbitMQ這個(gè)技巧是關(guān)于連接管理的。生產(chǎn)環(huán)境一定要用連接池管理Connection不要每次發(fā)消息都新建連接。Connection的創(chuàng)建是重操作頻繁創(chuàng)建會(huì)打滿文件描述符最終導(dǎo)致連接被拒絕。Channel也是類似同一個(gè)線程可以復(fù)用同一個(gè)Channel。Kafka的實(shí)戰(zhàn)技巧是關(guān)于Consumer的。消費(fèi)者啟動(dòng)后要注意設(shè)置一個(gè)合理的max.poll.interval.ms默認(rèn)值是5分鐘。如果你的單條消息處理時(shí)間超過(guò)5分鐘消費(fèi)者會(huì)被判定為假死觸發(fā)rebalance消息會(huì)被其他消費(fèi)者重新消費(fèi)極容易引發(fā)重復(fù)消費(fèi)。遇到過(guò)好幾次莫名重復(fù)消費(fèi)的線上事故排查到最后都是這個(gè)參數(shù)的問(wèn)題。RocketMQ的實(shí)戰(zhàn)技巧是關(guān)于消息軌跡的。如果線上消息莫名其妙就沒(méi)了先在控制臺(tái)的消息軌跡里查一下消息的完整生命周期看是生產(chǎn)端沒(méi)發(fā)出去、broker丟失還是消費(fèi)端拉到了但處理失敗。RocketMQ的軌跡功能非常完善定位問(wèn)題效率很高。不要一上來(lái)就去翻服務(wù)日志那樣太慢。最后再講一個(gè)所有消息隊(duì)列都適用的排查思路消息隊(duì)列的問(wèn)題本質(zhì)上是鏈路問(wèn)題。排查時(shí)一定不要只看單個(gè)節(jié)點(diǎn)要從生產(chǎn)者、broker、消費(fèi)者三段同時(shí)看哪一段有瓶頸數(shù)據(jù)會(huì)說(shuō)話的。把消息隊(duì)列的指標(biāo)監(jiān)控做起來(lái)比什么大牛排查技巧都管用。