談一談消息隊列
本文轉載自微信公眾號「SH的全棧筆記」,作者SH。轉載本文請聯系SH的全棧筆記公眾號。
本篇文章聊聊消息隊列相關的東西,內容局限于我們?yōu)槭裁匆孟㈥犃?,消息隊列究竟解決了什么問題,消息隊列的選型。
為了更容易的理解消息隊列,我們首先通過一個開發(fā)場景來切入。
不使用消息隊列的場景
首先,我們假設A同學負責訂單系統(tǒng)的開發(fā),B、C同學負責開發(fā)積分系統(tǒng)、倉儲系統(tǒng)。我們知道,在一般的購物電商平臺上,我們下單完成后,積分系統(tǒng)會給下單的用戶增加積分,然后倉儲系統(tǒng)會按照下單時填寫的信息,發(fā)出用戶購買的商品。
那問題來了,積分系統(tǒng)、倉儲系統(tǒng)如何感知到用戶的下單操作?
你可能會說,當然是訂單系統(tǒng)在創(chuàng)建完訂單之后調用積分系統(tǒng)、倉儲系統(tǒng)的接口了
OK,直接調用接口的方式在目前來看沒有什么問題。于是B、C同學就找到A同學,說讓他在訂單完成后,調用一下他們的接口來通知一下積分系統(tǒng)和倉儲系統(tǒng),來給用戶增加積分、發(fā)貨。A同學想著,就這兩個系統(tǒng),應該還好,OK我給你加了。
但是隨著系統(tǒng)的迭代,需要感知訂單操作的系統(tǒng)也越來越多,從之前的積分系統(tǒng)、倉儲系統(tǒng)2個系統(tǒng),擴充到了5個。每個系統(tǒng)的負責同學都需要去找A同學,讓他人肉的把對應系統(tǒng)的通知接口加上。然后就因為加了這一個接口,又需要把訂單重新發(fā)布一遍。
這對A同學來說實際上是很痛苦的一件事情,因為A同學有自己的任務、排期,一有新系統(tǒng)就需要去添加通知接口,發(fā)布服務,會打亂A的開發(fā)計劃,增加開發(fā)量。同時還需要去梳理在開發(fā)期間,新增的代碼到底能不能夠上線。一旦不能上線,但是又沒有檢查到,上線就直接炸了
而且,如果5個系統(tǒng)如果有哪個需要額外的字段,或者是更新了接口什么的,都需要麻煩A同學修改。5個系統(tǒng)就這樣跟A系統(tǒng)強耦合在了一起。
除此之外,整個創(chuàng)建訂單的調用鏈因為同步調用這5個系統(tǒng)的通知接口而加長,這減慢了接口的響應速度,降低了用戶側的購物、下單體驗。前面的至少影響的還是內部的員工,但是現在直接是影響到了用戶,明顯是不可取的方案。
可以看到,整個的調用鏈路加長了,更別提,在同步調用中,如果其余的系統(tǒng)發(fā)生了錯誤,或者是調用其他系統(tǒng)的時候出現了網絡抖動,核心的下單流程就會被阻塞住,甚至會在下單的界面提示提示用戶出錯,整個的購物體驗又被拉低了一個檔次。更何況,在實際的業(yè)務中,調用鏈比這個長的多。
可能有人會說了, 這不就是個同步調用問題嘛?訂單系統(tǒng)的核心邏輯,我還是采用同步來處理,但是后續(xù)的通知我采用異步的方式,用線程池去處理,這樣調用鏈路不就恢復正常了?
就單純對于減少鏈路來說,的確可行。但是如果某一個流程失敗了呢?難道失敗就失敗了嗎?我下單成功了不漲積分?該給我發(fā)的貨甚至沒有發(fā)貨?這合理嗎?
同時,失敗了訂單系統(tǒng)是不是要去處理呢?否則因為其他的系統(tǒng)拉垮了整個主流程,誰還來你這買東西呢?
那有什么辦法,既能夠減少調用的鏈路,又能夠在發(fā)生錯誤的時候重試呢?歸根結底,核心思想就是像增加積分、返優(yōu)惠券的流程不應該和主流程耦和在一起,更不應該影響主流程。
試想,我們能不能在訂單系統(tǒng)完成自己的核心邏輯之后,把訂單創(chuàng)建的消息放到一個隊列中去,然后訂單系統(tǒng)就返回給用戶下單成功的結果了。然后其他的系統(tǒng)從這個隊列中收到了下單成功的消息,就各自的去執(zhí)行各自的操作,例如增加積分、返優(yōu)惠券等等操作。
后續(xù)如果有新的系統(tǒng)需要感知訂單創(chuàng)建的消息,直接去訂閱這個隊列,消費里面的消息就好了?這雖然跟真實的消息的隊列有些出入,但其思路是完成吻合的。
為什么需要消息隊列
通過上面的例子,我們大致就能夠理解為什么要引入消息隊列了,這里簡單總結一下。
異步
對于實時性不是很高的業(yè)務,例如給用戶發(fā)送短信、郵件通知,調用第三方的接口,都可以放到消息隊列里去。因為相對于核心訂單流程來說,短信、郵件晚一些發(fā)送,對用戶來說影響不是很大。同時還可以提升整個鏈路的響應時間。
削峰
假設我們有服務A,是個無狀態(tài)的服務。通過橫向擴展,它可以輕松抗住1w的并發(fā)量,但是這N個服務實例,底層訪問的都是同一個數據庫。數據庫能抗住的并發(fā)量是有限的,如果你的機器足夠好的話,可能能夠抗住5000的并發(fā),如果服務A的所有請求全部打向數據庫,會直接把數據打掛。
解耦
像上文舉的例子,訂單系統(tǒng)在創(chuàng)建了訂單之后需要通知其他的所有系統(tǒng),這樣一來就把訂單系統(tǒng)和其余的系統(tǒng)強耦合在了一起。后續(xù)的可維護性、擴展性都大大降低了。
而通過消息隊列來關聯所有系統(tǒng),可以達到解耦的目的。
像上圖這種模式,如果后續(xù)再有新系統(tǒng)需要感知訂單創(chuàng)建的消息,只需要去消費「訂單系統(tǒng)」發(fā)送到MQ中的消息即可。同樣,訂單系統(tǒng)如果需要感知其余系統(tǒng)的某些事件,也只是從MQ中消費即可。
通過MQ,達成服務之間的松耦合,服務內的高內聚,提升了服務的自治性。
消息隊列選型
已知的消息隊列有Kafka、RocketMQ、RabbitMQ和ActiveMQ。但是由于ActiveMQ現在用的公司比較少了,這里就不做討論,我們著重討論前三種。
Kafka
Kafka最初來自于LinkedIn,是用于做日志收集的工具,采用Java和Scala開發(fā)。其實那個時候已經有ActiveMQ了,但是在當時ActiveMQ沒有辦法滿足LinkedIn的需求,于是Kafka就應運而生。
在2010年底,Kakfa的0.7.0被開源到了Github上。到了2011年,由于Kafka非常受關注,被納入了Apache Incubator,所有想要成為Apache正式項目的外部項目,都必須要經過Incubator,翻譯過來就是孵化器。旨在將一些項目孵化成完全成熟的Apache開源項目。
你也可以把它想象成一個學校,所有想要成為Apache正式開源項目的外部項目都必須要進入Incubator學習,并且拿到畢業(yè)證,才能走入社會。于是在2012年,Kafka成功從Apache Incubator畢業(yè),正式成為Apache中的一員。
Kafka擁有很高的吞吐量,單機能夠抗下十幾w的并發(fā),而且寫入的性能也很高,能夠達到毫秒級別。但是有優(yōu)點就有缺點,能夠達到這么高的并發(fā)的代價是,可能會出現消息的丟失。至于具體的丟失場景,我們后續(xù)會討論。
所以一般Kafka都用于大數據的日志收集,這種日志丟個一兩條無傷大雅。
而且Kafka的功能較為簡單,就是簡單的接收生產者的消息,消費者從Kafka消費消息。
RabbitMQ
RabbitMQ是很多公司對于ActiveMQ的替代方法,現在仍然有很多公司在使用。其優(yōu)點在于能保證消息不丟失,同Kafka,天平往數據的可靠性方向傾斜必然導致其吞吐量下降。其吞吐量只能夠達到幾萬,比起Kafka的十萬吞吐來說,的確是較低的。如果遇到需要支撐特別高并發(fā)的情況,RabbitMQ可能會無法勝任。
但是RabbitMQ有比Kafka更多的高級特性,例如消息重試和死信隊列,而且寫入的延遲能夠降低到微妙級,這也是RabbitMQ一大特點。
但RabbitMQ還有一個致命的弱點,其開發(fā)語言為Erlang,現在國內精通Erlang的人不多,社區(qū)也不怎么活躍。這也就導致可能公司內沒有人能夠去閱讀Erlang的源碼,更別說基于其源碼進行二次開發(fā)或者排查問題了。所以就存在RabbitMQ出了問題可能公司里沒人能夠兜的住,維護成本非常的高。
之所以有中小型公司還在使用,是覺得其不會面臨高并發(fā)的場景,RabbitMQ的功能已經完全夠用了。
RocketMQ
RocketMQ來自阿里,同Kakfa一樣也是從Apache Incubator出來的頂級項目,用Java語言進行開發(fā),單機吞吐量和Kafka一樣,也是十w量級。
RocketMQ的前身是阿里的MetaQ項目,2012年在淘寶內部大量的使用,在阿里內部迭代到3.0版本之后,將MetaQ的核心功能抽離出來,就有了RocketMQ。RocketMQ整合了Kafka和RabbitMQ的優(yōu)點,例如較高的吞吐量和通過參數配置能夠做到消息絕對不丟失。
其底層的設計參考了Kafka,具有低延遲、高性能、高可用的特點。不同于Kafka的單一日志收集功能,RocketMQ被廣泛運用于訂單、交易、計算、消息推送、binlog分發(fā)等場景。
之所以能夠被運用到多種場景,這要歸功于RocketMQ提供的豐富的功能。例如延遲消息、事務消息、消息回溯、死信隊列等等。
- 延遲消息 就是不會立即消費的消息,例如某個活動開始前15分鐘提醒用戶這樣的場景
- 事務消息 其主要解決數據庫事務和MQ消息的數據一致性,例如用戶下單,先發(fā)送消息到MQ,積分增加了,但是訂單系統(tǒng)在發(fā)出消息之后掛了。這樣用戶并沒有下單成功,但是積分卻增加了,明顯是不符合預期的
- 消息回溯 顧名思義,就是去消費某個Topic下某段時間的歷史消息
- 死信隊列 沒有被正常消費的消息,首先會按照RocketMQ的重試機制重試,當達到了最大的重試次數之后,如果消費仍然失敗,RocketMQ不會立即丟掉這條消息,而是會把消息放入死信隊列中。放入死信隊列的消息會在3天后過期,所以需要及時的處理
消息隊列會丟消息嗎
在不使用消息隊列的場景中,我們吹了很多消息隊列的優(yōu)點,但同時也提到了消息隊列可能會丟失消息,我們也可以通過參數的配置來使消息絕對不丟失。
那消息是在什么情況下丟失的呢?消息隊列中的角色可以分為3類,分別是生產者、MQ和消費者。一條消息在整個的傳輸鏈路中需要經過如下的流程。
生產者將消息發(fā)送給MQ,MQ接收到這條消息后會將消息存儲到磁盤上,消費者來消費的時候就會把消息返給消費者。先給出結論,在這3種場景下,消息都有可能會丟失。
接下來我們一步一步來分析一下。
生產者發(fā)送消息給MQ
生產者在發(fā)送消息的過程中,由于某些意外的情況例如網絡抖動等,導致本次網絡通信失敗,消息并沒有被發(fā)送給MQ。
MQ存儲消息
當MQ接收到了來自生產者的消息之后,還沒有來得及處理,MQ就突然宕機,此時該消息也會丟失。
即使MQ開始處理消息,并且將該消息寫入了磁盤,消息仍然可能會丟失。因為現代的操作系統(tǒng)都會有自己的OS Cache,因為和磁盤交互是一件代價相當大的事情,所以當我們寫入文件的時候會先將數據寫入OS Cache中,然后由OS調度,根據策略觸發(fā)真正的I/O操作,將數據刷入磁盤。
而在刷入磁盤之前,MQ如果宕機,在OS Cache中的數據就會全部丟失。
消費者消費消息
當消息順利的經歷了生產者、MQ之后,消費者拉取到了這條消息,但是當其還沒來得及處理的時候,消費者突然宕機了,這條消息就丟了(當然你如果沒有提交Offset的話,重啟之后仍然可以消費到這條消息)
原來我們以為用上了消息隊列,就萬無一失了,沒想到逐步分析下來能有這么多坑。任何一個步驟出錯都有可能導致消息丟失。那既然這樣,上文提到的可以通過參數配置來實現消息不會丟失是怎么一回事呢?
這里我們不去聊具體的MQ是如何實現的,我們來聊聊消息零丟失的實現思路。
消息最終一致性方案
涉及到的系統(tǒng)有訂單系統(tǒng)、MQ和積分系統(tǒng),訂單系統(tǒng)為生產者,積分系統(tǒng)為消費者。
首先訂單系統(tǒng)發(fā)送一個訂單創(chuàng)建的消息給MQ,該消息的狀態(tài)為Prepare狀態(tài),狀態(tài)為Prepare狀態(tài)的消息不會被消費者給消費到,所以可以放心的發(fā)送。
然后訂單系統(tǒng)開始執(zhí)行自身的核心邏輯,你可能會說,訂單系統(tǒng)本身的邏輯執(zhí)行失敗了怎么辦?剛剛的prepare消息不就成了臟數據?實際上在訂單系統(tǒng)的事務失敗之后,就會觸發(fā)回滾操作,就會向MQ發(fā)送消息,將該條狀態(tài)為Prepare的數據給刪除。
訂單系統(tǒng)核心事務成功之后,就會發(fā)送消息給MQ,將狀態(tài)為prepare的消息更新為commit。沒錯,這就是2PC,一個保證分布式事務數據一致性的協(xié)議。
眼尖的你可能發(fā)現了一個問題,我發(fā)送了prepare消息之后,還沒來得及執(zhí)行本地事務,訂單系統(tǒng)就掛了怎么辦?此時訂單系統(tǒng)即使重啟也不會向MQ發(fā)送刪除操作,這個prepare消息不就是一直存在MQ中了?
先給出結論,不會。
如果訂單系統(tǒng)發(fā)送了prepare消息給MQ之后自己就宕機了,MQ確實會存在一條不會被commit的數據。MQ為了解決這個問題,會定時輪詢所有prepare的消息,跟對應的系統(tǒng)溝通,這條prepare消息是要進行重試還是回滾。所以prepare消息不會一直存在于MQ中。這樣一來,就保證了消息對于生產者的DB事務和MQ中消息的數據一致性。
再來看一種更加極端的情況,假設訂單系統(tǒng)本地事務執(zhí)行成功之后,發(fā)送了commit消息到MQ,此時MQ突然掛了。導致MQ沒有收到該commit消息,在MQ中該消息仍然處于prepare狀態(tài),這怎么辦?
同樣的,依賴于MQ的輪詢機制和訂單系統(tǒng)溝通,訂單系統(tǒng)會告訴MQ這個事務已經完成了,MQ就會將這條消息設置成commit,消費者就可以消費到該消息了。
接下來的流程就是消息被消費者消費了,如果消費者消費消息的時候本地事務失敗了,則會進行重試,再次嘗試消費這條消息。