導讀
講述消息系統在現代化演進中軟硬一體化，百萬隊列，分級存儲等諸多競争力特性的誕生和落地效果。探讨業界領先的 Shared-Log 存儲計算分離，FFM與協程，RDMA 傳輸，列式存儲等技術，将消息向流的領域延伸。
1970年代末，消息系統用于管理多主機的列印作業，這種削峰解耦的能力逐漸被标準化為“點對點模型”和稍複雜的“釋出訂閱模型”，實作了資料處理的分布式協同。随着時代的發展，Kafka，Amazon SQS，RocketMQ，Azure Service Bus，Google Pub/Sub，RabbitMQ等衆多産品為開發者在不同業務場景下提供了富有競争力的解決方案，并擴充出順序，事務，定時消息，精确一次投遞等豐富的語義和特性，讓消息系統成為分布式系統中的标準元件。
消息系統伴随着雲原生與微服務理念的成長，見證了Serverless和事件驅動這樣高度動态和異步的系統強大的生命力，而消息隊列本身的架構和模型也随之改變。希望和大家分享一下消息系統這樣的PaaS / MaaS 層基礎設施服務，在現代化演進中是如何享受軟硬體發展紅利的，如何滿足Stream Processing和Real-Time Analysis場景下日益增長的資料分析需要和平衡成本與機器效能之間的沖突。讓消息基礎架構真正擁抱雲原生，結合近年來火熱的Shared-Log存儲計算分離來降本，配合FFM，協程，使用者态TCP和RDMA傳輸，列式存儲等技術進一步将消息向流的領域延伸，成為未來EDA，全面Serverless和輕計算的底座。
一、消息的發送和接收
1.1. 發送，公平還是貪心
從使用者或者用戶端視角看，消息的發送看似隻是通過rpc的方式，将序列化後的資料傳輸給消息隊列服務端并處理響應。在分布式環境中要考慮的細節較多，例如：
1.發送延遲：消息系統服務端一般會跨可用區部署來提供容災能力，如何就近發送減少延遲和流量費用？
2.多語言支援：生産者相比于在雲上規模化部署的消費者而言，來源更加廣泛，從小小的單片機，五花八門的前端頁面，到複雜的後端服務内部通信，對多語言多協定的支援有着更強烈的訴求。
3.多場景支援：需要适配多種模型，例如同步binlog的順序消息場景，物聯網海量用戶端的場景等。
4.發送失敗：由于服務端當機或網絡問題發送失敗，用戶端行為是什麼？是選擇資料反壓還是快速失敗。
圖：用戶端請求直接傳輸到存儲節點Broker和經過Proxy的差別
當我們回望網絡速度不夠快，“移動資料” 不如 “移動計算” 的時代，往往會提及經典的直連網絡架構。典型産品有HDFS，Kafka，Redis和Elasticsearch，用戶端和多個服務端節點之間直接建立連接配接，相比于所有流量先經過Proxy處理的架構，有效降低延遲，提高實時性。事實上用戶端不得不對服務端的各種分布式問題進行容錯，産生了更複雜的服務發現和負載均衡機制，用戶端還需要知道如何 “優雅的” 處理單點故障，這些都加大了用戶端版本滾動更新的難度。
而通過Proxy模式實作存儲計算分離，還可以在Proxy實作請求緩存，共享鑒權和服務發現等中繼資料。這種職責的劃分也能顯著簡化跨地域組網，異地多活等高階能力的配置。在有了Proxy之後，多語言用戶端接入的問題也被簡化，隻需要在無狀态的Proxy解析多種協定即可。至于多了一條通信延遲的問題，Proxy在服務端内部和後端存儲叢集之間可以使用性能更高的私有協定，配合FlatBuffer這樣低序列化開銷的庫以及RDMA通信等技術的發展，不會有顯著差異。
圖：TCP 擁塞控制 [2]
現代的消息隊列朝着更 “聰明的” 錯誤處理能力不斷努力着，并提供使用者選擇政策的機會。從網絡層面看發送失敗，本質上這是一種不可達，也可以認為是一種擁塞現象。一般有兩種解決方案：TCP-BBR的擁塞控制技術和類似銳速的多倍發包算法。在一些偏日志收集型的應用裡，消息隊列會更加關注全局吞吐，選擇類似TCP-BBR 這樣基于溢水原理的算法，用術語說是找到一個最合适的BDP（帶寬延遲積），當服務端繁忙時通過反壓來降低用戶端的發送速度，例如Kafka的攢批機制 (accumulate，參數為linger.ms) 和Nagle算法的思想類似，都是通過将小包攢為大包提升性能，也有在架構層實作滑動視窗和Flink基于信任值的流控這樣的方式，來更優雅的處理反壓。
與之對應的，銳速采用了多倍發包來最大化利用帶寬，事實上消息隊列也會提供快速失敗的政策，由使用者決定錯誤處理行為，是重試更多次還是走兜底鍊路。即将發送消息的逾時時間配置較短，通過快速多次重試來重發消息，這是一種全局非公平的貪心政策。多用于像RocketMQ這樣更加注重資料重要性和實時性的消息隊列裡 (服務端異步寫延遲<1ms，同步寫或多副本延遲為幾毫秒)，在存儲層繁忙時選擇 “快速失敗” 而非排隊等待，另一個角度也可以說是服務端希望用更高優先級去滿足那些線上應用的需求。
1.2. 消費，記錄多種模型的狀态
消息的消費，本質上是一個服務端配合下的兩階段送出，有一定分布式基礎的工程師會這樣描述 “服務端有一些邏輯或實體上的隊列，用戶端使用長輪詢對服務端發起請求，有消息則傳回，沒有消息時請求會在服務端挂起。消息在應用中進行業務邏輯的處理後，向服務端報告消費結果。” 服務端的使命就是更好更快的滿足這些嗷嗷待哺的用戶端們，記錄隊列中目前拉取的位點，維護取走的消息句柄和消息量。對于消息隊列的存儲層來說，需要精心設計資料結構來維護這些内容，并使用Paxos/Raft這樣的一緻性協定在服務端内傳播來保持高可用，它們也被稱為服務端的 “狀态”。
我們先來看一些簡單的例子：例如Kafka用内置Topic記錄Group的位點送出，複用消息多副本的複制鍊路保證可靠性，友善位點的監控與回溯。RocketMQ中訂閱組數量相比 Kafka高出兩三個數量級，是以選擇定時對記憶體位點做Checkpoint。對于一個特定的訂閱組和一個隊列，僅需要一個位點數字就可以描述消費進度，隊列模型在絕大多數場景下工作的簡單高效，但也有一些天生的缺陷：
1.消費者按照隊列次元負載均衡存在前提與假設：
a.隊列數不均等導緻負載不均，例如8個隊列3個消費者，最佳配置設定是3，3，2。
b.該模型假設了各個用戶端能力均等，實際生産中新舊機型混部，無法充分利用計算能力。
2.隊列中有慢任務會阻塞整個隊列。例如有位點為34567的5條消息，消費offset = 5 時業務邏輯耗時非常久，并發消費模式下67兩條消息消費較快，而觀察到的堆積一直為3造成誤判。
3.消費者或者服務端當機，業務對産生幾秒的消費重複依然敏感，影響使用者體驗，例如短信推送場景。
甚至，我們還有更有代表性的場景來命中這些 “缺陷”，例如渲染業務，隊列中每一條消息代表一個渲染任務。
1.消費者數量較多，同一個訂閱組可能有成百上千台機器同時消費。
2.該場景下單條資料的處理耗時較長，需要幾秒至幾個小時不等。
3.由于消費者負載高和大量使用競價執行個體，導緻消費方程序假死和當機率遠高于一般業務。
傳統的消息隊列一般會采用和Kafka類似的隊列模型，就會遇到很經典的 “Work-Stealing” 難題，任務的負載無法均衡的配置設定到所有消費方，單條消息的阻塞會影響後續消費成功消息位點的送出。此時我們想要的是一個基于不可見時間的投遞算法，該算法大緻的工作流程如下：
1.用戶端設定一個不可見時間，例如5分鐘，并向服務端拉取一批消息。
2.服務端傳回一批消息，并在背景開始倒計時5分鐘，消息上會附加一個字段用來辨別，也稱為handle。
3.如果用戶端5分鐘内沒有送出消費成功（ack by handle），5分鐘後用戶端再次可以擷取到這批消息。
很快我們就會發現這個模型還是有缺陷的，假如消費者拉取消息1分鐘後立刻當機了，業務不得不忍受4分鐘的延遲才能再次處理，哪怕此時其他消費者還是空閑狀态。這個時候就可以選擇将消息的不可見時間設定為1分鐘，在用戶端處理業務的同時不停的refresh不可見時間，例如每隔30秒就調用change invisible time，使剩餘的不可見時間更新為1分鐘，此時無論用戶端何時當機，消息的延遲時間會控制在1分鐘之内。
在RocketMQ中，這種基于區間和單條消息進行消費的方式被稱為 “pop消費”，對應的用戶端實作是SimpleConsumer，它的簡單性在于用戶端不再需要關心複雜的負載均衡和位點管理，也更容易适配多語言。這種管理能力在Pulsar中會更加複雜，被稱為WorkQueue模式和Range Ack位點管理。當服務端做的越多，用戶端和使用者擔心的事就越少，帶來了極大的靈活性。這種業務模型的演進驅動了消息雲存儲模型的變遷，當然這一切也是有代價的，例如SimpleConsumer這樣的無狀态消費模式來說，消息拉取的平均耗時要高于常用的PullConsumer，服務端用戶端互動的次數也會更多。
二、服務端能力提升
這些用戶端接口和政策的豐富，依托于服務端技術的提升。AIO，零拷貝，DirectIO 這些技術逐漸普及，極大地簡化了建構高性能系統的複雜度，設計合理的單機存儲引擎通常能夠達到每秒處理十萬甚至百萬級别的寫入性能。使用者的關注點也從單純的讀寫吞吐能力轉向了産品的功能特性，畢竟吞吐量可以通過水準擴容來解決，而增加一套單獨的元件，維護難度指數上升。作為使用者，總是期望可以盡可能的減少外部依賴，不會僅僅因為産品有一個精緻的大盤頁面而選擇一款産品，但可能會由于産品的功能缺失而被迫放棄，例如金融級産品強依賴傳輸和存儲加密能力。
不同于開放的社群，對于雲廠商來說消息隊列中很重要的核心競争力就是建構 “統一的消息核心”，在其上适配多種産品的接入協定，為所有産品提供一緻的底層能力，來最大化功能複用的收益。這種情況下，每适配一個新的産品，所付出的邊際成本是遞減的，這也導緻NoSQL類的資料庫，消息隊列，緩存元件，甚至日志服務都逐漸走向一個融合的生态。我認為現代消息隊列對于存儲特性的富化，主要會展現在以下幾個方面：海量隊列支援，分級存儲降成本，随着分布式檔案系統的成熟而進行Layered Replication 架構演進産生多模存儲的形态，多副本政策的靈活調整，以及更好更快的支援流式任務。
2.1. 海量隊列與多模統一
不同消息産品的側重點有差異，個人了解Kafka其實更側重于全局吞吐量，RocketMQ更偏向于實時性的應用，而RabbitMQ通過消息的模型去承載了一定的業務邏輯，MQTT場景則需要支援海量的裝置和Topic。這些差異化特性和競争力是消息隊列兩種領域模型的擴充，即基于主題的釋出訂閱模型和基于隊列的點對點模型，而 “統一的消息核心” 需要很好的适配多模的場景。
衆所周知，社群版本的Kafka會對每個分區建構獨立的LogSegment來存儲消息，配合檔案的磁盤空間預配置設定等政策，在海量隊列的場景下，存在明顯的性能問題。RocketMQ中消息以CommitLog的形式混合存儲，為了保證前台寫入性能，将所有的資料使混合存儲，曆史版本中也曾出現過使用IO線程中做零拷貝，結果大量的缺頁中斷導緻線程阻塞的問題。在存儲引擎層面不得不進行大量的定制優化，使用單獨的冷熱分離服務進行權重計算，使用單獨的冷讀線程池等。
但索引仍然是獨立的小檔案，預設每個消費隊列的檔案存儲30萬條消息索引，一個索引占用20個位元組，這樣每個索引檔案的大小是300 * 1000 * 20 / 1024 / 1024 ≈ 5.72M，百萬個隊列會占用數 TB 的磁盤空間。對于隊列數量極多的業務場景下，必然選擇将這些索引也混合存儲，在檔案系統層面合并為大檔案。我們發現，類似RocksDB這樣支援排序的LSM結構，能夠合并小檔案批量寫入SST，顯著改進了大量小檔案的碎片化問題。性能測試表明，使用RocksDB存儲索引替代使用原生檔案版索引的情況下，單機可以支援百萬級别的隊列數，在單機4萬隊列數 (含重試隊列) 的場景下本地磁盤的索引空間占用從200G降低到30G，相比于使用檔案版cq作為索引，cpu開銷增加10%左右。
那麼類LSM的結構的真正優勢在什麼地方？衆所周知，存儲引擎有in-place updates（原地更新）和out-of-place updates（異地更新）兩種結構。原地更新中最典型的結構是B樹及其變種，而B樹是一種平衡多路搜尋樹，在每個葉子中插入多個條目來降低樹的高度，提升查詢的性能和穩定性。由于B樹的連續性結構（索引結構能夠為新插入的索引條目根據鍵值最終整理排序，與所有其他已經存在的條目放置在一起），從B樹讀入的資料不太可能在其在緩沖區中的短時間内再次被引用以進行第二次插入，是以在B樹中寫請求無法像LSM做成組送出。
第二，通常存儲引擎持久化的資料量遠超記憶體，對于任何可合并操作（比如說按照隊列進行順序通路）的通路率來說，都是朝着冷資料的方向發展的，而在LSM結構中Compaction機制會有一個局部聚簇的批處理效應，即連續取回資料能較好的配合預讀，這是一個極其顯著的優勢。誠然Compaction機制在資料庫領域工作的非常好，但在消息隊列場景下大Value的特殊性，額外的讀寫放大開銷對消息系統來是非常高的，也會嘗試使用fast 論文中類似WiscKey的一些KV分離特性減少讀寫放大。Compaction機制也可以使用類似TerarkDB和Titan提出的方案來優化，這也增加了消息系統實作 “Topic粒度的 TTL” 和 “位點周期性訂正” 的業務特性的複雜度。
2.2. 分級存儲轉化資料資産
近些年來，如何控制好不斷膨脹的基礎設施成本，在社群裡是一個熱議的話題，在商業産品可以轉化為價格競争力上，是吸引使用者從VM自模組化式轉向商業産品的重要賣點之一。過去消息系統基于本地磁盤或者塊存儲（雲盤）建構存儲，雲SSD磁盤的價格在0.8-1 元/GB/月，而對象存儲的價格普遍在0.1-0.12元/GB/月，在成本上有數倍的差距。生産中單機的存儲空間使用量為2TB，每個月的存儲成本為1600元，使用對象存儲存儲冷資料，假設熱資料占比20%，存儲成本為每月200 * 0.8 + 1800 * 0.1 = 340元。為什麼不進一步壓縮塊存儲的容量，做到幾乎極緻的成本呢？事實上，在分級存儲的場景下，一味的追求過小本地磁盤容量帶來的 “邊際效益” 是遞減的。
主要有以下原因：
- 故障備援，消息隊列作為基礎設施中重要的一環，穩定性高于一切。雖然對象存儲本身可用性較高，如果遇到網絡波動等問題時，使用對象存儲作為主存儲，非常容易産生反壓導緻熱資料無法寫入，而熱資料讀寫影響線上生産業務，這對于業務可用性的影響是緻命的。
- 過小的本地磁盤，在價格上沒有明顯的優勢。衆所周知，雲計算是注重普惠和公平的，使用分層存儲後計算成本占比上升，熱資料的寫入流量不變。如果選用50G左右的塊存儲，又需要等價200G的ESSD級别的塊存儲能提供的IOPS讀寫能力，則其機關成本幾乎是低IOPS的普通塊存儲的數倍。
- 上傳時能夠更好的使用本地磁盤 “攢批” 減少對象存儲的請求費用。
- 讀取時能夠對“溫熱” 資料提供更低的延遲和節約讀取成本，對取回的冷資料也可以放置在本地磁盤緩存。
“誰離資料最近，誰就越有可能成功” 是資料驅動決策的核心要點，消息隊列作為應用開發的資料通道，受限于本地磁盤的容量和成本限制，往往隻能存儲幾天。分級存儲提供了大幅延長消息的生命周期的低成本方案，在轉冷時可以靈活動态的使用 FlatBuffer，Parquet等多種資料格式，将消息隊列從通道轉化為使用者的資料資産的存儲池，這些技術進一步将消息向流的領域延伸，成為未來EDA和輕計算的底座。
2.3. 基于分布式存儲降計算成本
分級存儲能夠有效解決冷資料存儲成本問題，并不能很好的降低整體擁有成本的。消息系統内部對于熱資料使用了多副本技術，一方面是為了保證資料的可靠性和服務的高可用，同時副本均可讀能夠提供了更大的讀取帶寬，适配一寫多讀的消息場景。這種架構一些場景下不可避免的會有如下問題：
- 寫性能不夠極緻，消息隊列為了降低用戶端複雜度，往往采用Y型寫，服務端内部的Raft或者其他日志複制的算法占用了較多帶寬，帶來寫性能和吞吐的下降。每次消息資料更新的時候，需要先更新主副本，通過一緻性協定進行複制和 Quorum計算，寫入時延至少在四跳網絡RT（用戶端->主，主->備複制，備響應進度，主傳回成功），且會有長尾。
- 計算成本浪費，備機一般不需要處理大量用戶端的讀請求，CPU使用率通常為主的一半。當主副本就能完全滿足讀寫請求時，備副本的計算能力就會浪費。這一點也很難通過使用混部，單程序多容器化（例如Flink的Slot機制，RocketMQ的 Broker Container機制）的等技術優化。
- 擴容縮容慢，當出現熱點或者需要緊急擴容的時候，例如社群版Kafka需要進行資料複制，生效慢。
- 多副本需要管控面元件來進行身份裁決，而許多團隊可能會害怕維護像ZK這樣的服務，擔心他出各種問題。而Kafka引入Kraft，RocketMQ的Dledger同步 CommitLog，JRaft Controller也都具有一定的複雜度，增加了整個系統架構的運維負擔。
多副本架構面臨了衆多難題，不容易實作單調讀 (Monotonic Reads)，副本之間還可能出現ISR(In-sync Replicas) 不同步的問題。幸運的是，我們有精彩的理論指導。微軟于2008年發表的PacificA論文中，就提出了基于日志的系統架構及其實作資料一緻性複制的三個方案：
圖：來自微軟論文 PacificA: Replication in Log-Based Distributed Storage Systems [8]
1.日志複制（Log Replication），類似于Raft中描述的Replicated State Machine（複制狀态機），主伺服器（Primary）和備伺服器（Secondary）之間複制日志，每個節點照相同的順序執行相同的指令。
2.日志合并（Log Merge），主維護資料結構，備伺服器不保持記憶體中的資料結構，僅接收檢查點（checkpoint）和日志複制。當主伺服器發生故障時，備伺服器可以通過加載和回放日志來恢複狀态。
3.分層架構（Layered Replication），讓資料的一緻性複制直接交給底層HDFS這樣的分布式檔案系統處理。
圖: 根據PolarDB團隊優化，分層後主節點和隻讀節點之間的網絡傳輸量減少了98%。
圖檔來自Analyze the Technical Essentials of PolarDB at the Architecture Level
消息隊列是資料密集型和時延敏感型的存儲應用，在分層架構中，寫延遲得到了充分的優化，無論是Kafka的sendfile syscal還是RocketMQ的mmap，由于核心機制的一些問題，無法充分利用現代硬體。而分布式檔案系統通常選擇基于SPDK這樣的使用者态檔案系統，Run-To-Complete線程模型和星型的2-3異步寫入，保證資料可靠性的同時，寫性能遠超雲盤（雲盤底層也是基于分布式檔案系統）和一些本地的舊 SSD。分層架構下，計算和記憶體資源的使用更加彈性，計算能力支援獨立精細的管理，随着使用者的負載動态擴縮容，計算節點 “無身份綁定”，擴容速度極快。
資料的可靠性和讀寫性能被 “更專業的團隊” 來解決。當然原有的一些技術也會随之改變，對于資料的讀取不再能夠依賴作業系統的pagecache，這一點業内也有很多漂亮的解決方案，例如PolarDB共享buffer Pool和WrapStream提出的 “分布式mmap” 都能很好的利用多個節點上的記憶體。
現代應用層存儲引擎踐行 “Log is Streaming” 的理念，将存儲的複雜度徹底解除安裝到更底層的分布式存儲。因為任何時候人力資源總是有限的，應當适時進行“減法”處理。每一項技術的引入都會帶來額外的複雜度，要避免過度工程化導緻的維護成本攀升。以功能特性極為豐富的Flink為例，自運維社群版的複雜度吓退了很多中小型使用者，保持存儲引擎的低依賴會讓産品擁有更廣泛的開發者群體，做到可持續發展。
2.4. 自閉環的流式處理能力
多年來像RocketMQ這樣的消息隊列一直被用于交易、供應鍊履約等核心鍊路，流轉着大量高價值的業務資料。Kafka等日志型消息産品也累積着大量的使用者行為資料，為了由此社群提出了一些解決方案讓 “資料資産” 通過計算産生價值，例如KStream 和KsqlDB這樣的輕計算解決方案，我也熟悉Spark和Flink這樣功能特性更加強大的平台，在我參與貢獻了社群基于FLIP 27/191新版Flink-Connector-RocketMQ的改進後。我意識到，現代消息隊列正朝着 “消息，事件，流” 一體的平台成長與發展。
Streaming Processing場景之是以複雜，是因為要在性能、正确性、代價上取平衡。
- 可重複讀，即資料的可回放性，不能産生“讀擺動”，以便在發生故障時可以從上一次成功處理的狀态（如快照）正确恢複。這意味着消息消費者（計算架構中一般稱為source）必須能夠在發生當機或網絡抖動等問題時，準确地定位并回放讀取特定位點的消息。而消息存儲系統必須保證下遊消費者觀察到的資料進度不會超出叢集已确認送出的最新狀态，即quorum原則下多數派的高水位，來確定消費過程的幂等性。
- 分區政策與時間水印，一個工程上常見的需求是在資料源保持分區數穩定，可以避免因為分區變化引發的負載不平衡。這個模型其實有很多隐式語義，例如服務端高可用機制完善，各個分區沒有熱點和資料傾斜的情況，下遊消費者所在的節點能力對等，以及消息隊列對于周期性的時間水印支援等等。在無需感覺分區數量模型下，如Google 的 Pub/Sub等服務，為了做到端到端的 “精确一次” 計算，則需要在計算側維護大量的handle狀态。在DSL機制中，使用者需要為accumulations和撤回retractions提供對應的實作，知道怎樣修正結果，例如join派生成新的table，那麼修正會從目前子拓撲往後傳播，這也造成了較高的端到端延遲。
- 資料流轉的IO開銷。即便是在現代的百G網絡環境下，流處理架構中的廣播（broadcast）、聚合（aggregation）、資料重新分區（shuffle）等反複讀寫消息隊列的操作有很長的IO耗時。而Flink Sink中TwoPhaseCommit等保持分布式事務的做法事實上也很複雜，給消息隊列帶來 “讀寫放大”。
計算架構和消息隊列存儲的分開維護，我們獲得了将計算邏輯與資料存儲解耦的靈活性，同時也增加了系統的維護難度和複雜性。未來，消息隊列的存儲層在其核心功能之上，也會內建一些輕量級的計算能力，如資料轉換、滾動聚合、以及支援視窗等概念和操作。目前消息隊列僅支援了一些簡單的Tag過濾，SQL92過濾的動作，如果存儲層引入schema的概念，能夠根據具體需要取回對應部分的資料，就就能夠進一步減少讀寫操作。
三、行業前沿探索
除了使用者群體龐大的RocketMQ和Kafka等消息産品，我們也關注到細分市場中的挑戰者，他們瞄準了行業痛點或者深挖性能，創造差異化的競争力，典型的有WrapStream和Redpanda。
3.1. 徹底去除本地磁盤
Kafka節點當機恢複時會涉及到資料的複制，流程複雜且耗時非常久。原因在于 Kafka強依賴本地磁盤，即使社群提出并實作了分級存儲等KIP，熱資料需要在多個本地磁盤上至少需要保留12-24小時，成本高。WarpStream選擇與Kafka做協定相容，完全去除對EBS的依賴，将Kafka直接建構在了對象存儲S3上。架構上分為類似Proxy角色的Agent和支援百萬TPS的Meta中繼資料管理服務。其核心處理流程如下：
1.發送消息，Agent将不同Topic的資料混合攢批寫入對象存儲，成功後讓Meta服務定序。
a.由于需要攢批，節省調用費和對象存儲本身的延遲較高的原因，導緻寫請求延遲很高，達到400ms。
b.對于多可用區場景，寫負載均衡和切流，是通過hack用戶端ID實作，不破壞原生Kafka協定。
c.對于幂等消息的支援，是通過offset由Meta服務裁決，并忽略沒有成功傳回的請求。這種設計在盤古的seal and new chunk調用，各種分級存儲append模型下的 fast rolling設計是一緻的。
2.接收消息，官方稱為分布式mmap，多個Agent構成了一緻性哈希到環，對相同分區的讀被聚集到同一個Agent上以提高緩存命中率。同時背景執行Compation提高回放速度，解決TTL問題。這種設計完全無狀态，元件水準擴容難度低，實作多租和大叢集的實作相對簡單。缺陷則是操作會過多依賴Meta服務。
另外官網也提到了一些低延遲改進的技術，例如對于Express One Zone這樣單可用區版本的對象存儲：
- 減小上傳buffer和逾時時間換性能。
- 支援寫多個bucket來實作類似2-3寫的技術，利用quorum原則來快速ack發送。
3.2. 用原生語言重構
Redpanda 選擇充分利用現代硬體特性，以原生語言實作低延遲和低雲成本為賣點，特别是對于發送長尾的改善。Redpanda的節點依賴了改進後的Raft（including an optimistic approach to Raft and parallel commits），冷資料依賴對象存儲。Kafka的挑戰者很多，大部分廠商會選擇以LogSegment作為切面，以減少Kafka計算層協定演進帶來的相容性問題。而Redpanda則選擇自底向上的重構，讓單個固定線程來執行對于單個分區的所有操作，包括網絡輪詢，異步IO，擷取事件和排程計算任務等，這種線程模型稱為thread-per-core，也稱為run-to-complete。Actor模型是一種強大的并發模型，縮小了臨界區的範圍，取代了Reactor下通過互斥鎖實作的多線程，預期所有操作可以在500 us以内處理完成，而使用C++開發帶來的确定性延遲能夠有效的降低基于JVM應用的長尾，帶來可預測的P99延遲。
第二條路線是對于熱資料轉冷後的維護，是由每個分區的Leader負責上傳，複用Raft鍊路複制中繼資料。
- 上傳的元件scheduler_service和archival_metadata_stm維護了一個類PID Controller的公平排程器算法，通過計算需要上傳到對象存儲的資料總量，并動态更新優先級。對于積壓量較大的分區，則優先級會更高。如果上傳積壓較小，則上傳過程的優先級會較低以減少背景流量對前台讀寫的幹擾。
- 資料取回時remote_partition和cache_service負責從對象存儲下載下傳資料并緩存，根據消費這請求的分區和位點計算出底層對應的混合日志段的相對偏移（hydrated log segment），然後預取并緩存以減少對對象存儲的調用次數，減小平均RT，也支援一些就近讀取政策減少跨可用區的流量以降低費用。
- 一些難點和提高性能的設計，例如Seastar (Using Boost.Beast）開發http用戶端提高對對象存儲通路的性能。資料上傳和緩存的管理需要考慮公平性，每個分區内部的狀态管理也很複雜。
對于Native語言的優劣勢，我們一直期望服務端能夠 “Write Once，Run EveryWhere”，實際上由于基礎設施想要充分利用好現代硬體的能力，不得不嵌入一些JNI或做一些指令集優化來提升熱點函數的性能。Redpanda特别提到對于Arm的支援，由于項目完全采用C++開發，對于依賴庫需要打開一些開關進行動态編譯，這個操作并沒有想象中跨平台那麼複雜。最終的結論是Arm比x86能夠帶來20%左右的成本下降，和我們讓RocketMQ嘗試從x86移植到Arm的收益是接近的。
3.3. 新技術引入與嘗試
現代消息隊列除了擴充場景以外，也在緻力于嘗試新興技術和軟硬一體化，以進一步提高性能降低成本。例如：
- 通信層：計算層Proxy和存儲節點使用TCP進行通信，而TCP通信存在一定的延遲，同時在高密容器模式部署時也存在帶寬限制。TCP協定是為廣域網而設計的，不能夠很好的支援資料中心的場景，我們嘗試将這條鍊路改為RDMA通信。RDMA技術可以讓應用程式直接通路遠端主機記憶體，在不涉及到網絡軟體棧的情況下。資料能夠被直接發送到緩沖區或者能夠直接從緩沖區裡接收，而不需要被複制到網絡層。可以直接在使用者态執行資料傳輸，不需要在核心态與使用者态之間做上下文切換，通路遠端主機記憶體而不消耗遠端主機中的任何CPU。并且将網絡協定棧的很多處理操作通過硬體offload，最終降低了網絡資料傳輸端到端的時延，保證了消息實作持久化存儲、吞吐量大且實時性高。從實際測試結果來看，能減少8%左右的CPU。當然，對于沒有用cpu set模式進行部署的Java應用來說，會導緻更多的請求長尾。
- 計算層：消息隊列自身也在嘗試引入JDK17的協程技術來改進大量異步操作帶來的代碼可維護性的問題。很多傳統優化項依然沒有達到極緻的狀态，例如大家所熟知的基于引用計數減少buffer的拷貝，分析一些熱點并定向JNI優化，或通過Native語言重構。例如在x86和Arm的不同架構下，SpinLock性能存在較大差異，講這些小功能優化為平台相關動态連結庫并引入，能夠以較低的代價獲得較大的性能提升。又比如對于一些重複的操作可以用JDK21中的FFM和SIMD技術進行優化，顯著的降低CPU開銷。
- 存儲層：在消息隊列的場景中，相同Topic下消息 Payload 存在着非常大的資料相關性，典型的壓縮比可以達到10:1。随着消息向流的領域延伸，我們也嘗試在消息的存儲格式中引入FlatBuffer和Parquet這樣的記憶體友好，反序列化開銷低的存儲格式來提升查詢性能。
參考文獻
1.John K. Ousterhout, et al. "Cloud Programming Simplified: A Berkeley View on Serverless Computing." arXiv preprint arXiv:1902.03383v1 (2019). https://arxiv.org/abs/1902.03383v1
2.TCP Slow-Start and Congestion Avoidance, https://commons.wikimedia.org/w/index.php?title=File:TCP_Slow-Start_and_Congestion_Avoidance.svg
3.Asterios Katsifodimos, et al. "Consistency and Completeness: Rethinking Distributed Stream Processing in Apache Kafka." In Proceedings of the ACM SIGMOD International Conference on Management of Data (SIGMOD '21). Association for Computing Machinery, 2021.
4.Flink, Apache. "Stateful computations over data streams." Accessed: Apr 23 (2021): 2021.
5.State management in Apache Flink®: consistent stateful distributed stream processing. https://dl.acm.org/doi/10.14778/3137765.3137777
6.Michael Armbrust, et al. "Delta Lake: High-performance ACID Table Storage over Cloud Object Stores." Databricks White Paper, Aug. 2020. https://www.databricks.com/wp-content/uploads/2020/08/p975-armbrust.pdf
7.Wei Cao, Zhenjun Liu, Peng Wang, Sen Chen, Caifeng Zhu, Song Zheng, Yuhui Wang, and Guoqing Ma. 2018. PolarFS: an ultra-low latency and failure resilient distributed file system for shared storage cloud database. Proc. VLDB Endow. 11, 12 (August 2018), 1849–1862. https://doi.org/10.14778/3229863.3229872
8.PacificA: Replication in Log-Based Distributed Storage Systems, https://www.microsoft.com/en-us/research/publication/pacifica-replication-in-log-based-distributed-storage-systems/
9.Heidi Howard and Richard Mortier. 2020. Paxos vs Raft: have we reached consensus on distributed consensus? In Proceedings of the 7th Workshop on Principles and Practice of Consistency for Distributed Data (PaPoC '20). Association for Computing Machinery, New York, NY, USA, Article 8, 1–9. https://doi.org/10.1145/3380787.3393681
10.Abhishek Verma, etc. 2015. Large-scale cluster management at Google with Borg. In Proceedings of the Tenth European Conference on Computer Systems (EuroSys '15). Association for Computing Machinery, New York, NY, USA, Article 18, 1–17. https://doi.org/10.1145/2741948.2741964
11.Peter A. Alsberg and John D. Day. 1976. A principle for resilient sharing of distributed resources. In Proceedings of the 2nd international conference on Software engineering (ICSE '76). IEEE Computer Society Press, Washington, DC, USA, 562–570. https://dl.acm.org/doi/10.5555/800253.807732
12.Sanjay Ghemawat, Howard Gobioff, and Shun-Tak Leung. 2003. The Google file system. SIGOPS Oper. Syst. Rev. 37, 5 (December 2003), 29–43. https://doi.org/10.1145/1165389.945450
13.K. Mani Chandy and Leslie Lamport. 1985. Distributed snapshots: determining global states of distributed systems. ACM Trans. Comput. Syst. 3, 1 (Feb. 1985), 63–75. https://doi.org/10.1145/214451.214456
作者:斜陽
來源-微信公衆号:阿裡雲開發者
出處:https://mp.weixin.qq.com/s/M02YiF9altrsODORb9gZuw