可靠消息隊列淺談

可靠消息隊列淺談

@招牌瘋子

綜述

消息隊列系統是大型分布式系統中常見的組成部分之一，目前市面上也已經出現了大量非常優秀的消息隊列或者具有消息隊列特征的數據流系統，它們各自有各自的特點，卻也同樣會有自己的不足，在某些特定的應用場景下，既有的消息隊列用起來總是欠那么點火候，這也是為什么redis作者也在寫一個全新的消息隊列系統的原因。本文的寫作目的，不是為了討論市面上已有消息隊列的優缺點，更不是為造輪子洗地，而只是從一個實際需求出發，完整地記錄我是如何從零開始實現一個自己用著順手的消息隊列系統的過程。如果我所做的這個東西剛好戳中了你的痛點，歡迎嘗試使用并提出自己的意見和建議。

目標

新浪和微博有大量系統嚴重依賴于我們老大@stvchu 所寫的memcacheQ，mcq 已經在大量線上業務中服務了好多年，其性能和穩定性無需置疑。我們組開發維護的圖片存儲和視頻存儲系統，也都是依靠mcq進行消息同步的。在多年的使用中，漸漸暴露出來一些可靠性和易用性上的不足，于是我們計劃開發一款全新的消息隊列，既保留mcq的優點，又能滿足一些新的需求。

1. 可靠性

   目前的mcq是單點的，一旦一臺mcq服務器故障，所有隊列的消息都將丟失。由于mcq非常穩定，目前并沒有這樣的事故發生，盡管如此，我們還是希望能夠從根本上杜絕這種情況，開發一個可靠性非常高的消息隊列，無需為單點問題擔憂。

   </li>

2. 多條消費者隊列

   在mcq中，一個消息入隊之后，消費者只能從隊列里取出來一次，在某些場景下，一個消息可能需要被多個消費者系統消費，比如微博圖片傳上來之后，需要被壓縮系統拿來進行壓縮，同時還要被分類系統拿去進行分類，還會被審核系統拿去審查，等等。目前的做法是，一個消費者先從隊列里拿到消息，處理完之后再重新入隊，這樣其他系統就可以繼續拿出來消費。弊端是將一個并行的任務生生做成了串行，而且更嚴重的是，一旦某個消費者系統拿了消息之后發生故障，未能將消息塞回來，那么后續的消費者就無法處理這個消息了。

   因此我們的一個目標是支持多個消費者隊列，它們之間互不影響，各自消費各自隊列里的消息。

   </li>

3. 消息確認和重新入隊

   消息確認也是保證消息可靠性的一個重要方面，目前mcq并不支持消息確認機制。試想一個消息被某個消費者拿走之后，還沒來得及處理，這個消費者系統就掛了，那么這個消息就丟失了，任何其他消費者都無法對它進行處理。之前看到別的部門同事介紹這種問題的處理方案，他們自己又開發了一個模塊，從mcq里拿到消息之后如果長時間沒有處理好，就重新寫入mcq中，也是挺不方便的。

   所以我們的新系統中，必須支持消息確認機制，未確認的消息一段時間之后將會自動重新進入隊列中，無需使用者操心。

   </li> </ol>

   開發過程

   接到這個項目是去年11月份，本以為只是一個簡單的東西很快就能搞完，沒想到最后越搞越復雜，一直到春節過后才找到靠譜的方案，最近才開發完成。

   1. 基于raft協議的可靠性消息隊列Express

      為了達到上面列出的第一個目標，我們選擇了在消息隊列系統中引入raft一致性協議進行消息同步，簡單地說，我們的系統一次起N個實例，客戶端可以連到任意一個實例上進行入隊和出隊操作，只要集群中有N/2+1個節點存活且確認拿到消息，即可認為消息處理成功；即使有少部分實例掛掉，系統依然可以對外提供服務；當掛掉的這些實例重啟或者新加入幾個實例之后，它們也可以自動同步到最新的狀態，然后繼續對外服務。

      在數據庫領域談可用性、一致性之類的問題已經談了很多年，大家都覺得raft已經是論文論證過，工業可用，而且比paxos簡單一萬倍的東西，應該是很好實現的，包括我自己也這么想。然而理想很豐滿，現實很骨干，消息隊列的應用場景跟數據庫是完全不同的，引入raft一個最大的問題就是處理能力嚴重不足！

      MQ這種東西，本來就是解決生產者和消費者速度不匹配的問題而誕生的，那么MQ系統一個最最基本的要求就是寫入速度必須要快，哪怕出隊速度慢點也無所謂，因為業務高峰期持續時間是有限的，高峰結束之后有的是時間讓消費者慢慢消化，更別說簡單粗暴多加幾臺消費者就好了。而一旦引入raft一致性協議，每個消息都要等半數以上的express實例確認之后才能返回成功，延時非常之高！要達到50ms左右（這個延時可以通過設置raft heartbeat來降低，但同時會提高系統負載，效果不明顯）。

      再加上，MQ系統在隊列堵塞的時候，短時間內累積的消息數量非常之多，以峰值10億條，每條消息200字節算，如果存在內存里，需要200GB的內存，對我們公司來說太奢者了，所以必須落地存儲。而raft協議同步的基礎是增量日志（WAL），這就導致每一個消息會帶來兩次磁盤寫入，大家都知道磁盤 I/O是非常慢的操作，更進一步降低了系統入隊速度。

      為了克服這個困難，我們一度改寫了raft庫對WAL的操作邏輯，將WAL同時作為我們MQ的落地存儲模塊，依靠記錄log ID之類的方法，改寫了raft生成snapshot時的操作流程，使得一個消息寫入只會帶來一次磁盤I/O。雖然提高了一點點性能，但是開發成本實在高得離譜。

      再后來我們發現導致延時高更重要的原因是多個實例之間同步消息過程。舉例來說，集群入隊一個消息需要50ms的話，實例之間通過raft庫相互同步和確認就需要花掉40ms以上，與這樣的消耗相比，上面說到的兩次磁盤寫操作根本不值一提。正在這個時候，etcd發布了它們的首個正式版本，并在博客中大肆吹噓了一番他們自己開發的raft庫，由于goraft優化無望，我們開始嘗試使用etcd的raft庫重新進行開發。

      這個過程按下不表，總之找到了一些捷徑之后很快就實現了MQ的業務邏輯，進行測試之后性能也確實有較大改觀，看來goraft真的是不行。考慮到上面分析的性能瓶頸所在的結論，和對etcd/raft進行改造帶來的巨大工作量，我們改回了雙寫落地的方案，也并沒有慢多少。

      到此時，關于可靠性這個目標已經實現了。但是不管怎么說，raft所帶來的延時是無法避免的，即使是etcd本身，處理能力也就1000qps而已。最終結果就是，express只能用于我們自己的業務線上，因為我們更看重可靠/可用性，性能要求不是特別高。

      </li>

   2. Topic（話題）和Line（消費線路）

      為了支持多條消費者隊列，在express中，存在topic和line這兩個概念。一個topic可以擁有多條line，所有的消息入隊的目標是 topic，只需要入隊一次；而消息出隊只能從line里拿，每個line都有自己的名字，代表這條消費線路的具體用途，以圖片上傳舉例：

      • 所有消息入隊到名叫wb_img_upload的topic中
      • 壓圖模塊從名叫img_to_compress的line中拿屬于自己的消息并進行壓縮
      • 同時，分類模塊從名叫img_to_analysis的line中拿屬于自己的消息并進行分類
      • 其他各種模塊比如審核、統計等等，也從各自的line中拿消息，它們之間互不影響
      </ul>

      這種設計類似于nsq的topic和channel，但是nsq中為了使多個channel之間相互獨立，采用復制消息的方案，也就是說，入隊的消息，有幾個channel就得多復制幾次。而在express中，line里面其實只是存了消息的游標，實際消息還是存在topic中的，入隊也只需要寫入一次即可。

      </li>

      • 消息確認

        消息確認是針對line來說的，每條line在創建的時候就需要設定一個recycle（回收）時間，從line里拿走一個消息時，會同時得到該消息的ID，當你處理完之后，需要明確地告訴express以確認這條消息已經消費完畢，否則，超過recycle時間之后，該消息會重新加入到這條 line的消息隊列中，可以被其他消費者再次消費。由于重新入隊這個機制的存在，會打亂消息入隊的順序，甚至導致消息重復消費，從而引出時序性和冪等性的問題，這個我們在下一節進行探討。

        </li> </ol>

        時序性和冪等性

        消息隊列系統的時序性也是一個常見的問題了，在某些場景下，我們確實需要非常嚴格的按時序排列的消息，比如微博的feed，當然是需要先入隊的消息先展示，后入隊的消息后展示。那么如果MQ本身是嚴格時序的，當然是最好了（其實對MQ來說也是最簡單的了，比如mcq就是），但問題是，你可以保證MQ 按時間順序出隊，卻無法保證消費者按時間順序消費完畢啊。注意我說的是消費完畢，畢竟每個消費者遇到的情況都是不同的，總有干的快的也總有干的慢的。解決辦法也很簡單，在消息體內包含消息生成的時間即可，所有消息在需要展示的時候才按生成時間再排序，即可嚴格時序輸出。

        那么，實際上來說就是，真正保證時序性的，并不是消息隊列，而是消費者！

        除非有一種情況，消費者只有一個，所有出隊的消息都由它來展示，那在消費者這一層面就不可能打亂消息順序，只需要MQ按時序出隊即可。以我目前貧瘠的知識量來看，這種場景是小作坊產品才會遇到的，根本不可能出現在需要解決流量高峰問題的互聯網產品中；再者，如果只需要一個消費者就能順序處理的話，你還有必要引入一個消息隊列嗎？當然，由于視野所限，這一段內容不一定正確，總之核心思想就是這種只有一個消費者的場景太罕見了，直接忽略掉算了。

        接上面的結論繼續說，由于嚴格時序性需要靠消費者這一層來保證，那么對于MQ來說，只需要在一定程度上保證時序性即可。為了能夠實現消息確認和重入來保證消息不丟失，是不可能同時保證嚴格時序性的。事實上，redis作者Antirez在介紹他的disque時，也基本上是這么說的。

        接下來說冪等性，MQ里的冪等性指的是一個消息如果被消費者多次消費，對于消費者來說效果應該是一樣的。比如，用戶上傳了一張圖片，這個圖片需要被壓圖模塊壓縮成縮略圖，壓圖服務器A取到了消息并壓圖完畢，但是在跟express確認消息的時候由于網絡故障沒成功，過了一段時間之后express讓這條消息重新入隊，壓圖服務器B又拿到了這個消息并壓圖完畢，這個時候，對于整個壓圖模塊來說，是沒有影響的，因為B壓好的圖，文件名和路徑跟A之前壓好的是完全一樣的，只是把那個圖覆蓋了而已，用戶依然可以看到，這就叫冪等消費者。

        在某些消費者邏輯中，消息不是冪等的，重復處理同一個消息會造成數據錯亂。這種情況下就不適合依靠MQ的消息重入機制來保證消息不丟了，而因該消費者自己制定策略。那么在express中創建line的時候，只需要設置recycle為空即可。

        這兒再插入一塊關于丟消息的內容。在我們的系統中，消息一旦被確認入隊，就不可能丟失，因為入隊時已經落地存儲完成，只可能多次出隊，這樣的設計是為了最大程度保證消息不丟。而有些MQ并不能保證不丟消息，比如NSQ，有一部分消息是在內存里的，如果這時候NSQ宕機，消息就沒了。這是速度和安全取舍的問題，并無優劣之分。

        簡化版本UQ

        上面一大篇密密麻麻的文字，主要記錄了我設計和實現一款基于raft的，非常可靠的消息隊列系統express的過程。相信我們遇到的問題應該也戳中了很多同學的痛點。但是老實來說，保證了可靠性但損失了性能的express并不適合大家使用，因此我把express中關于隊列的邏輯拆出來，只保留那些好用的功能，犧牲了高可用這一部分，做成一個新的持久化消息隊列開源出來，取名UQ，希望有需要的朋友可以拿去試用:

        https://github.com/buaazp/uq

        UQ的具體功能和用法在github上有詳細的介紹，本文中就不浪費篇幅了，只列出我認為能吸引到你的幾個亮點：

        • 支持topic、line
        • 支持消息確認和重新入隊
        • 支持超多協議，memcached, redis, http RESTful
        • 支持多種存儲后端，goleveldb, memory
        • 支持集群以提高可用和吞吐，通過etcd同步topic和line
        • 不錯的性能
        • 可以用來替代memcacheQ
        </ul>

        更多的信息請訪問github查看，我這人不喜歡拉票，但是如果你覺得UQ還不錯，請記得給我加個星。

        最后

        以上這些東西算是對我近幾個月來的工作做一個小的總結，俗話說，站在巨人的肩膀上，你自然就看得遠，致謝名單如下：

        • 首先是我的領導@馬健將，真正低調的大牛，express的設計碰壁的時候全靠小馬哥幫我撥亂反正
        • 然后是我的更大的領導@stvchu，更是低調的大牛，還能說啥，mcq作者，在隊列功能的設計上給我直擊要害的點撥
        • 最后是好基友@Xscape，在時間緊迫的畢業找工作期間還抽空幫我修改語法錯誤
        </ul>

        送給我一直看不慣的嘴炮們，寫代碼真的不容易，嘴上說起來都是頭頭是道，真能寫出來的才算數。

        玩得開心！

        來自：http://blog.buaa.us/talk-about-mq/