面試官：在MQ的整個消息生產(chǎn)消費過程中，如何保障消息100%被消費？

候選人：MQ有個ACK機制，確保消息100%被消費。

面試官：好吧，可以回去等通知了……

這道面試題在考察MQ組件時算是老生常談了，不知道你是如何回答的？

我們平時都在使用MQ，但使用技術框架只是第一步，去弄明白它的底層原理、深挖技術真相，才是每一位IT從業(yè)者的基操。

這里說明一點，想要回答好面試官的問題，最好還是要有金字塔思維——金字塔思維就是從不同維度上來思考問題的一種方式，不重不漏，集體窮盡。

MQ作為異步通訊的消息中間件，其功能除了解耦生產(chǎn)者與消費者，還能用于大流量的削峰填谷，解決業(yè)務的最終一致性問題，那么消息的“可靠性”就顯得尤為重要了，比如說商品出庫后的庫存數(shù)據(jù)通過MQ同步到財務系統(tǒng)，如果消息的可靠性沒有保障，那財務系統(tǒng)的存貨成本分析數(shù)據(jù)就無法有效支撐財務團隊。

準確來說，我們需要保障MQ消息的可靠性，需要從三個層面/維度解決：生產(chǎn)者100%投遞、MQ持久化、消費者100%消費，這里的100%消費指的是消息不少消費，也不多消費。

由于MQ是基礎網(wǎng)絡通訊的中間件，網(wǎng)絡通訊必然因丟包、網(wǎng)絡抖動等原因產(chǎn)生數(shù)據(jù)丟失，MQ組件本身也會由于宕機或軟件崩潰而中止服務，從而造成數(shù)據(jù)丟失，那么我們就需要從這兩個根本原因著手補償，這里科普一下RabbitMQ和Kafka是怎么解決的。

RabbitMQ

這里我必須先提一提RabbitMQ的消息協(xié)議——AMQP（Advanced Message Queuing Protocol，高級消息隊列協(xié)議），在面試時我經(jīng)常問候選人一個問題：RabbitMQ用的是什么消息協(xié)議？大部分候選人是回答不出來AMQP的，更不用說AMQP模型是如何設計的了。 在服務器中，三個主要功能模塊連接成一個處理鏈完成預期的功能：

Exchange：接收發(fā)布應用程序發(fā)送的消息，并根據(jù)一定的規(guī)則將這些消息路由到消息隊列

Queue：存儲消息，直到這些消息被消費者安全處理完為止

Binding：定義了exchange和queue之間的關聯(lián)，提供路由規(guī)則

使用這個模型我們可以很容易地模擬出存儲轉(zhuǎn)發(fā)隊列和主題訂閱這些典型的消息中間件概念。 接下來我們看看RabbitMQ的消息確認機制是如何保障消息可靠性的。一、生產(chǎn)者端

通過API將信道（channel）設置為confirm模式，則每條消息會被分配一個唯—ID

如果消息投遞成功，也就是說消息已經(jīng)到達broker了，信道會發(fā)送ack給生產(chǎn)者，回調(diào)ConfirmCallback接口，帶上唯一ID

如果發(fā)生錯誤導致消息丟失，比如通過某個RoutingKey無法路由到某個Queue，則會發(fā)送nack給生產(chǎn)者，回調(diào)ReturnCallback接口，并帶上唯一ID和異常信息

ack和nack只有一個被觸發(fā)，只觸發(fā)一次，而且是異步執(zhí)行，意味著生產(chǎn)者不需要等待，可以繼續(xù)發(fā)送新消息

二、消費者端

聲明隊列時，指定noack=false， 表示消費者不會自動提交ack，broker會等待消費者手動返回ack、才會刪除消息，否則立刻刪除

broker的ack沒有超時機制，只會判斷鏈接是否斷開，如果斷開了(比如消費者處理消息過程中宕機)，消息會被重新發(fā)送，所以消費者要做好消息冪等性處理

此外，RabbitMQ除了消息確認機制，還有另一種方式——使用事務消息：消息生產(chǎn)端發(fā)送commit命令，MQ同步返回commit ok命令，這種方式由于需要同步阻塞等待MQ返回是否投遞成功，才能執(zhí)行別的操作，性能較差，因此不推薦使用。三、MQ本身通常來說，消息是在內(nèi)存中存儲通訊的，而基于內(nèi)存的都是會有數(shù)據(jù)丟失的問題產(chǎn)生，服務一重啟，數(shù)據(jù)就隨之銷毀。 在RabbitMQ中對數(shù)據(jù)的持久化有三方面：交換機持久化、隊列持久化、消息持久化。

交換機持久化：exchange_declare創(chuàng)建交換機時通過參數(shù)durable=true指定，如：channel.exchangeDeclare(exchangeName, “direct/topic/header/fanout”,true);第三個參數(shù)就是設置durable值

隊列持久化：queue_declare創(chuàng)建隊列時通過參數(shù)durable=true指定，如：channel.queueDeclare("queue.persistent.name",true, false, false, null)，第二個參數(shù)就是設置durable值

消息持久化：new AMPQMessage創(chuàng)建消息時通過參數(shù)指定，如：channel.basicPublish("exchange.persistent", "persistent",MessageProperties.PERSISTENT_TEXT_PLAIN, "persistent_test_message".getBytes())，或者設置參數(shù)deliveryMode=2來指定：AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();builder.deliveryMode(2);

上面只是說了API層的實現(xiàn)，那RabbitMQ底層又是怎么做消息持久化的呢？如果指定了持久化參數(shù)，它們會以append的方式寫文件，會根據(jù)文件大?。J16M）自動切割，生成新的文件，RabbitMQ啟動時會創(chuàng)建兩個進程，一個負責持久化消息的存儲，另一個負責非持久化消息的存儲（當內(nèi)存不夠時會用到）。 消息存儲時，會在一個叫ets的表中記錄消息在文件中的映射以及相關信息（包括ID、偏移量、有效數(shù)據(jù)、左邊文件、右邊文件），消息讀取時根據(jù)該信息到文件中讀取，同時更新信息。 消息刪除時只從ets刪除，變?yōu)槔鴶?shù)據(jù)，當垃圾數(shù)據(jù)超出比例（默認50%），并且文件數(shù)達到3個，就會觸發(fā)垃圾回收：鎖定左右兩個文件，整理左邊文件有效數(shù)據(jù)、將左邊文件有效數(shù)據(jù)寫入左邊，更新文件信息，刪除右邊，完成合并；當一個文件的有效數(shù)據(jù)等于0時，刪除該文件。 寫入文件前先寫入buffer緩沖區(qū)，如果buffer已滿，則寫入文件，注意，此時只是操作系統(tǒng)的頁存，還沒落盤。 每隔25ms刷一次磁盤（比如Linux中的fsync命令），不管buffer（fd的讀、寫緩存區(qū)）滿沒滿，都將buffer和頁存中的數(shù)據(jù)落盤。 還有另外一種落盤機制：每次消息寫入后，如果沒有后續(xù)寫入請求，則直接刷盤。 KafkaKafka在MQ領域以性能高、吞吐能力強、消息堆積能力強等等優(yōu)勢稱著，常常用于日志收集、消息系統(tǒng)、用戶活動跟蹤、運營指標、流式處理等等場景，講之前先簡單聊聊Kafka的架構設計：

Consumer Group：消費者組，消費者組內(nèi)每個消費者負責消費不同分區(qū)的數(shù)據(jù)，提高消費能力，這是邏輯上的一個訂閱者。

Topic：可以理解為一個隊列，Topic將消息分類，生產(chǎn)者和消費者面向的是同一個Topic。

Partition：為了實現(xiàn)擴展性，提高并發(fā)能力，一個Topic以多個Partition的方式分布到多個Broker上，每個Partition是一個有序的隊列，一個Topic的每個Partition都有若干個副本（Replica），一個Leader和若干個Follower；生產(chǎn)者發(fā)送數(shù)據(jù)的對象，以及消費者消費數(shù)據(jù)的對象，都是通過Leader，F(xiàn)ollower負責實時從Leader中同步數(shù)據(jù)，保持和Leader數(shù)據(jù)的同步；當Leader發(fā)生故障時，某個Follower還會成為新的Leader。

一、生產(chǎn)者端Kafka消息發(fā)送端有個ACK機制。

設置ack參數(shù)：ack=0，表示不重試，Kafka不需要返回ack，極有可能各種原因造成丟失；ack=1，表示Leader寫入成功就返回ack了，F(xiàn)ollower不一定同步成功；ack=all或ack=-1，表示ISR列表中的所有Follower同步完成再返回ack。

設置參數(shù)unclean.leader.election.enable: false，禁止選舉ISR以外的Follower為Leader，只能從ISR列表中的節(jié)點中選舉Leader；可能會犧牲Kafka的可用性，但是能夠提高消息的可靠性。

重試機制，設置tries > 1，表示消息重發(fā)次數(shù)。

設置最小同步副本數(shù)min.insync.replicas > 1，沒滿足該值前，Kafka不提供讀寫服務，寫操作會異常。

通過設置最小同步副本數(shù)和ACK機制，可以讓MQ在性能與可靠性上達到平衡。二、消費者端手工提交offset（偏移量）：Kafka消費者在拉取消息后，默認會自動提交offset，由于消費者每次都會根據(jù)offset來消費消息的，如果消費者處理業(yè)務失敗，實際上我們是要重新消費的，所以我們要在消息處理成功后再手工提交offset，確認消息能夠成功消費。 同樣地，消費者的業(yè)務代碼也要做好冪等性校驗。三、MQ本身很簡單，通過減小broker刷盤間隔來實現(xiàn)高可靠。 要深究其原理，得從Kafka的持久化機制來看。

磁盤的順序讀寫：與RabbitMQ不同，Kafka是基于磁盤讀寫的，那為什么Kafka的吞吐量還這么大呢？原因是Kafka的讀寫是用順序讀寫的，不需要尋址隨機讀寫，而由于是用磁盤來寫數(shù)據(jù)，消息堆積能力必然比內(nèi)存型的RabbitMQ更強

利用了操作系統(tǒng)的零拷貝技術：避免CPU將數(shù)據(jù)從一塊存儲拷貝到另外一塊存儲，關于零拷貝這里不詳述，與Java應用不同，Kafka的消息不需要在用戶緩沖區(qū)處理磁盤數(shù)據(jù)再返回，所以才能用零拷貝技術

分區(qū)分段+索引：Kafka的消息實際上分布存儲在一個一個小的segment中的，每次文件讀寫也是直接操作segment，為了進一步優(yōu)化查詢，Kafka又默認為分段后的數(shù)據(jù)文件建立了索引文件（就是文件系統(tǒng)上的.index文件），這種分區(qū)分段+索引的設計，不僅提升了數(shù)據(jù)讀取的效率，同時也提高了數(shù)據(jù)操作的并行度（類似ConcurrentHashMap的分段鎖機制）。

批量壓縮&批量讀寫：多條消息一起壓縮進行傳輸（比如gzip格式）與讀寫，節(jié)省帶寬

直接操作page cache：雖然Kafka是Java寫的，也基于JVM運行，但Kafka的消息讀寫是直接操作操作系統(tǒng)頁存的，而不是在JVM的堆內(nèi)存，這樣就避免JVM的GC耗時及對象創(chuàng)建耗時，且讀寫速度更高，JVM進程重啟緩存也不會丟失

理解了Kafka的持久化機制是直接讀寫頁存+定時刷盤的方式，我們只需要設置刷盤策略即可在性能與可靠性上權衡。 Kafka提供3個參數(shù)來優(yōu)化刷盤機制：

log.flush.interval.messages //多少條消息刷盤1次

log.flush.interval.ms //隔多長時間刷盤1次

log.flush.scheduler.interval.ms //周期性的刷盤。

總結一下關于框架類的面試題，最重要是得掌握技術框架的底層實現(xiàn)原理、適用場景，基本上回答出這兩方面就OK了，其它奇奇怪怪的細節(jié)問題要是答不出來，咱就引導面試官說出自己對框架的理解即可，畢竟細節(jié)的問題太多了。那怎么才算掌握呢？起碼能通過框架的特性，根據(jù)需要實現(xiàn)一個簡易版本，比如說自己實現(xiàn)一個Spring框架、實現(xiàn)一個MQ組件等等。

由淺入深，化難為易。

審核編輯 ：李倩