《一篇文了解分布式隊列編程：從模型、實戰到優化》要點：
本文介紹了一篇文了解分布式隊列編程：從模型、實戰到優化，希望對您有用。如果有疑問，可以聯系我們。

本文由美團點評技術團隊出品,一篇文助你掌握分布式隊列編程的要義.從模型到實戰再到優化,基本涵蓋你可能踩的坑與其解決辦法.

作為一種基礎的抽象數據結構,隊列被廣泛應用在各類編程中.大數據時代對跨進程、跨機器的通訊提出了更高的要求,和以往相比,分布式隊列編程的運用幾乎已無處不在.但是,這種常見的基礎性的事物往往容易被忽視,使用者往往會忽視兩點：

使用分布式隊列的時候,沒有意識到它是隊列.

有具體需求的時候,忘記了分布式隊列的存在.

文章首先從最基礎的需求出發,詳細剖析分布式隊列編程模型的需求來源、定義、結構以及其變化多樣性.通過這一部分的講解,作者期望能在兩方面幫助讀者：一方面,提供一個系統性的思考方法,使讀者能夠將具體需求關聯到分布式隊列編程模型,具備進行分布式隊列架構的能力；另一方面,通過全方位的講解,讓讀者能夠快速識別工作中碰到的各種分布式隊列編程模型.

文章的第二部分實戰篇.根據作者在新美大實際工作經驗,給出了隊列式編程在分布式環境下的一些具體應用.這些例子的基礎模型并非首次出現在互聯網的文檔中,但是所有的例子都是按照挑戰、構思、架構三個步驟進行講解的.這種講解方式能給讀者一個“從需求出發去構架分布式隊列編程”的旅程.

老司機介紹

劉丁,新美大廣告平臺CRM系統技術負責人,曾就職于Amazon、Tripadvisor.2014年加入美團,先后負責美團推薦系統、智能篩選系統架構,作為技術負責人主導了美團廣告系統的開發和上線.目前致力于推進新美大廣告運營的標準化、自動化和智能化.

新美大廣告平臺是美團、大眾點評雙平臺的營銷推廣平臺,幫助商戶推廣店鋪品牌及提升客流量.

1、模型篇

模型篇從基礎的需求出發,去思考何時以及如何使用分布式隊列編程模型.建模環節非常重要,因為大部分中高級工程師面臨的都是具體的需求,接到需求后的第一個步驟就是建模.通過本篇的講解,希望讀者能夠建立起從需求到分布式隊列編程模型之間的橋梁.

何時選擇分布式隊列

通信是人們最基本的需求,同樣也是計算機最基本的需求.對于工程師而言,在編程和技術選型的時候,更容易進入大腦的概念是RPC、RESTful、Ajax、Kafka.在這些具體的概念后面,最本質的東西是“通訊”.

所以,大部分建模和架構都需要從“通信”這個基本概念開始.當確定系統之間有通訊需求的時候,工程師們需要做很多的決策和平衡,這直接影響工程師們是否會選擇分布式隊列編程模型作為架構.從這個角度出發,影響建模的因素有四個：When、Who、Where、How.

When：同步VS異步

通信的一個基本問題是：發出去的消息什么時候需要被接收到?這個問題引出了兩個基礎概念：“同步通訊”和“異步通訊”.根據理論抽象模型,同步通信和異步通信最本質的差別來自于時鐘機制的有無.同步通信的雙方需要一個校準的時鐘,異步通信的雙方不需要時鐘.

現實的情況是,沒有完全校準的時鐘,所以沒有絕對的同步通信.同樣,絕對異步通信意味著無法控制一個發出去的消息被接收到的時間點,無期限的等待一個消息顯然毫無實際意義.

所以,實際編程中所有的通信既不是“同步通信”也不是“異步通信”；或者說,既是“同步通信”也是“異步通信”.特別是對于應用層的通信,其底層架構可能既包含“同步機制”也包含“異步機制”.判斷“同步”和“異步”消息的標準問題太深,而不適合繼續展開.作者這里給一些啟發式的建議：

發出去的消息是否需要確認,如果不需要確認,更像是異步通信,這種通信有時候也稱為單向通信(One-WayCommunication).

如果需要確認,可以根據需要確認的時間長短進行判斷.時間長的更像是異步通信,時間短的更像是同步通信.當然時間長短的概念是純粹的主觀概念,不是客觀標準.

發出去的消息是否阻塞下一個指令的執行,如果阻塞,更像是同步,否則,更像是異步.

無論如何,工程師們不能生活在混沌之中,不做決定往往是最壞的決定.當分析一個通信需求或者進行通信構架的時候,工程師們被迫作出“同步”還是“異步”的決定.當決策的結論是“異步通信”的時候,分布式隊列編程模型就是一個備選項.

Who：發送者接收者解耦

在進行通信需求分析的時候,需要回答的另外一個基本問題是：消息的發送方是否關心誰來接收消息,或者反過來,消息接收方是否關心誰來發送消息.如果工程師的結論是：消息的發送方和接收方不關心對方是誰、以及在哪里,分布式隊列編程模型就是一個備選項.因為在這種場景下,分布式隊列架構所帶來的解耦能給系統架構帶來這些好處：

無論是發送方還是接收方,只需要跟消息中間件通信,接口統一.統一意味著降低開發成本.

在不影響性能的前提下,同一套消息中間件部署,可以被不同業務共享.共享意味著降低運維成本.

發送方或者接收方單方面的部署拓撲的變化不影響對應的另一方.解藕意味著靈活和可擴展.

Where：消息暫存機制

在進行通信發送方設計的時候,令工程師們苦惱的問題是：如果消息無法被迅速處理掉而產生堆積怎么辦、能否被直接拋棄?如果根據需求分析,確認存在消息積存,并且消息不應該被拋棄,就應該考慮分布式隊列編程模型構架,因為隊列可以暫存消息.

How：如何傳遞

對通信需求進行架構,一系列的基礎挑戰會迎面而來,這包括：

可用性,如何保障通信的高可用.

可靠性,如何保證消息被可靠地傳遞.

持久化,如何保證消息不會丟失.

吞吐量和響應時間.

跨平臺兼容性.

除非工程師對造輪子有足夠的興趣,并且有充足的時間,采用一個滿足各項指標的分布式隊列編程模型就是一個簡單的選擇.

分布式隊列編程定義

很難給出分布式隊列編程模型的精確定義,由于本文偏重于應用,作者并不打算完全參照某個標準的模型.總體而言：分布式隊列編程模型包含三類角色：發送者(Sender)、分布式隊列(Queue)、接收者(Receiver).發送者和接收者分別指的是生產消息和接收消息的應用程序或服務.

需要重點明確的概念是分布式隊列,它是提供以下功能的應用程序或服務：

接收“發送者”產生的消息實體；

傳輸、暫存該實體；

為“接收者”提供讀取該消息實體的功能.

特定的場景下,它當然可以是Kafka、RabbitMQ等消息中間件.但它的展現形式并不限于此,例如：

隊列可以是一張數據庫的表,發送者將消息寫入表,接收者從數據表里讀消息.

如果一個程序把數據寫入Redis等內存Cache里面,另一個程序從Cache里面讀取,緩存在這里就是一種分布式隊列.

流式編程里面的的數據流傳輸也是一種隊列.

典型的MVC(Model–view–controller)設計模式里面,如果Model的變化需要導致View的變化,也可以通過隊列進行傳輸.這里的分布式隊列可以是數據庫,也可以是某臺服務器上的一塊內存.

抽象模型

最基礎的分布式隊列編程抽象模型是點對點模型,其他抽象構架模型居于改基本模型上各角色的數量和交互變化所導致的不同拓撲圖.具體而言,不同數量的發送者、分布式隊列以及接收者組合形成了不同的分布式隊列編程模型.記住并理解典型的抽象模型結構對需求分析和建模而言至關重要,同時也會有助于學習和深入理解開源框架以及別人的代碼.

點對點模型(Point-to-point)

基礎模型中,只有一個發送者、一個接收者和一個分布式隊列.如下圖所示：

生產者消費者模型(Producer–consumer)

如果發送者和接收者都可以有多個部署實例,甚至不同的類型；但是共用同一個隊列,這就變成了標準的生產者消費者模型.在該模型,三個角色一般稱為生產者(Producer)、分布式隊列(Queue)、消費者(Consumer).

發布訂閱模型(PubSub)

如果只有一類發送者,發送者將產生的消息實體按照不同的主題(Topic)分發到不同的邏輯隊列.每種主題隊列對應于一類接收者.這就變成了典型的發布訂閱模型.在該模型,三個角色一般稱為發布者(Publisher),分布式隊列(Queue),訂閱者(Subscriber).

MVC模型

如果發送者和接收者存在于同一個實體中,但是共享一個分布式隊列.這就很像經典的MVC模型.

編程模型

為了讓讀者更好地理解分布式隊列編程模式概念,這里將其與一些容易混淆的概念做一些對比 .

分布式隊列模型編程和異步編程

分布式隊列編程模型的通訊機制一般是采用異步機制,但是它并不等同于異步編程.

首先,并非所有的異步編程都需要引入隊列的概念,例如：大部分的操作系統異步I/O操作都是通過硬件中斷( Hardware Interrupts)來實現的.

其次,異步編程并不一定需要跨進程,所以其應用場景并不一定是分布式環境.

最后,分布式隊列編程模型強調發送者、接收者和分布式隊列這三個角色共同組成的架構.這三種角色與異步編程沒有太多關聯.

分布式隊列模式編程和流式編程

隨著Spark Streaming,Apache Storm等流式框架的廣泛應用,流式編程成了當前非常流行的編程模式.但是本文所闡述的分布式隊列編程模型和流式編程并非同一概念.

首先,本文的隊列編程模式不依賴于任何框架,而流式編程是在具體的流式框架內的編程.

其次,分布式隊列編程模型是一個需求解決方案,關注如何根據實際需求進行分布式隊列編程建模.流式框架里的數據流一般都通過隊列傳遞,不過,流式編程的關注點比較聚焦,它關注如何從流式框架里獲取消息流,進行map、reduce、 join等轉型(Transformation)操作、生成新的數據流,最終進行匯總、統計.

2、實戰篇

這里所有的項目都是作者在新美大工作的真實案例.實戰篇的關注點是訓練建模思路,所以這些例子都按照挑戰、構思、架構三個步驟進行講解.受限于保密性要求,有些細節并未給出,但這些細節并不影響講解的完整性.

另一方面,特別具體的需求容易讓人費解,為了使講解更加順暢,作者也會采用一些更通俗易懂的例子.通過本篇的講解,希望和讀者一起去實踐“如何從需求出發去構架分布式隊列編程模型”.

需要聲明的是,這里的解決方案并不是所處場景的最優方案.但是,任何一個稍微復雜的問題,都沒有最優解決方案,更談不上唯一的解決方案.實際上,工程師每天所追尋的只是在滿足一定約束條件下的可行方案.當然不同的約束會導致不同的方案,約束的松弛度決定了工程師的可選方案的寬廣度.

信息采集處理

信息采集處理應用廣泛,例如：廣告計費、用戶行為收集等.作者碰到的具體項目是為廣告系統設計一套高可用的采集計費系統.

典型的廣告CPC、CPM計費原理是：收集用戶在客戶端或者網頁上的點擊和瀏覽行為,按照點擊和瀏覽進行計費.計費業務有如下典型特征：

• 采集者和處理者解耦,采集發生在客戶端,而計費發生在服務端.

• 計費與錢息息相關.

• 重復計費意味著災難.

• 計費是動態實時行為,需要接受預算約束,如果消耗超過預算,則廣告投放需要停止.
• 用戶的瀏覽和點擊量非常大.

挑戰

計費業務的典型特征給我們帶來了如下挑戰：

• 高吞吐量－－廣告的瀏覽和點擊量非常巨大,我們需要設計一個高吞吐量的采集架構.

• 高可用性－－計費信息的丟失意味著直接的金錢損失.任何處理服務器的崩潰不應該導致系統不可用.

• 高一致性要求－－計費是一個實時動態處理過程,但要受到預算的約束.收集到的瀏覽和點擊行為如果不能快速處理,可能會導致預算花超,或者點擊率預估不準確.所以采集到的信息應該在最短的時間內傳輸到計費中心進行計費.

• 完整性約束－－這包括反作弊規則,單個用戶行為不能重復計費等.這要求計費是一個集中行為而非分布式行為.
• 持久化要求－－計費信息需要持久化,避免因為機器崩潰而導致收集到的數據產生丟失.

構思

采集的高可用性意味著我們需要多臺服務器同時采集,為了避免單IDC故障,采集服務器需要部署在多IDC里面.

實現一個高可用、高吞吐量、高一致性的信息傳遞系統顯然是一個挑戰,為了控制項目開發成本,采用開源的消息中間件進行消息傳輸就成了必然選擇.

完整性約束要求集中進行計費,所以計費系統發生在核心IDC.

計費服務并不關心采集點在哪里,采集服務也并不關心誰進行計費.

根據以上構思,我們認為采集計費符合典型的“生產者消費者模型”.

架構

采集計費系統架構圖如下：

• 用戶點擊瀏覽收集服務(Click/View Collector)作為生產者部署在多個機房里,以提高收集服務可用性.

• 每個機房里采集到的數據通過消息隊列中間件發送到核心機房IDC_Master.

• Billing服務作為消費者部署在核心機房集中計費.

采用此架構,我們可以在如下方面做進一步優化：

• 提高可擴展性,如果一個Billing部署實例在性能上無法滿足要求,可以對采集的數據進行主題分區(Topic Partition)計費,即采用發布訂閱模式以提高可擴展性(Scalability).

• 全局排重和反作弊.采用集中計費架構解決了點擊瀏覽排重的問題,另一方面,這也給反作弊提供了全局信息.
• 提高計費系統的可用性.采用下文單例服務優化策略,在保障計費系統集中性的同時,提高計費系統可用性.

分布式緩存更新(Distributed Cache Replacement)

緩存是一個非常寬泛的概念,幾乎存在于系統各個層級.典型的緩存訪問流程如下：

• 接收到請求后,先讀取緩存,如果命中則返回結果.

• 如果緩存不命中,讀取DB或其它持久層服務,更新緩存并返回結果.

對于已經存入緩存的數據,其更新時機和更新頻率是一個經典問題,即緩存更新機制(Cache Replacement Algorithms ).典型的緩存更新機制包括：近期最少使用算法(LRU)、最不經常使用算法(LFU).

這兩種緩存更新機制的典型實現是：啟動一個后臺進程,定期清理最近沒有使用的,或者在一段時間內最少使用的數據.由于存在緩存驅逐機制,當一個請求在沒有命中緩存時,業務層需要從持久層中獲取信息并更新緩存,提高一致性.

挑戰

分布式緩存給緩存更新機制帶來了新的問題：

• 數據一致性低.分布式緩存中鍵值數量巨大,從而導致LRU或者LFU算法更新周期很長.在分布式緩存中,拿LRU算法舉例,其典型做法是為每個Key值設置一個生存時間(TTL),生存時間到期后將該鍵值從緩存中驅逐除去.考慮到分布式緩存中龐大的鍵值數量,生存時間往往會設置的比較長,這就導致緩存和持久層數據不一致時間很長.如果生存時間設置過短,大量請求無法命中緩存被迫讀取持久層,系統響應時間會急劇惡化.
• 新數據不可用.在很多場景下,由于分布式緩存和持久層的訪問性能相差太大,在緩存不命中的情況下,一些應用層服務不會嘗試讀取持久層,而直接返回空結果.漫長的緩存更新周期意味著新數據的可用性就被犧牲了.從統計的角度來講,新鍵值需要等待半個更新周期才會可用.

構思

根據上面的分析,分布式緩存需要解決的問題是：在保證讀取性能的前提下,盡可能地提高老數據的一致性和新數據的可用性.如果仍然假定最近被訪問的鍵值最有可能被再次訪問(這是LRU或者LFU成立的前提),鍵值每次被訪問后觸發一次異步更新就是提高可用性和一致性最早的時機.

無論是高性能要求還是業務解耦都要求緩存讀取和緩存更新分開,所以我們應該構建一個單獨的集中的緩存更新服務.集中進行緩存更新的另外一個好處來自于頻率控制.

由于在一段時間內,很多類型訪問鍵值的數量滿足高斯分布,短時間內重復對同一個鍵值進行更新Cache并不會帶來明顯的好處,甚至造成緩存性能的下降.通過控制同一鍵值的更新頻率可以大大緩解該問題,同時有利于提高整體數據的一致性,參見“排重優化”.

綜上所述,業務訪問方需要把請求鍵值快速傳輸給緩存更新方,它們之間不關心對方的業務.要快速、高性能地實現大量請求鍵值消息的傳輸,高性能分布式消息中間件就是一個可選項.這三方一起組成了一個典型的分布式隊列編程模型.

架構

如下圖,所有的業務請求方作為生產者,在返回業務代碼處理之前將請求鍵值寫入高性能隊列.Cache Updater作為消費者從隊列中讀取請求鍵值,將持久層中數據更新到緩存中.

采用此架構,我們可以在如下方面做進一步優化：

• 提高可擴展性,如果一個Cache Updater在性能上無法滿足要求,可以對鍵值進行主題分區(Topic Partition)進行并行緩存更新,即采用發布訂閱模式以提高可擴展性(Scalability).
• 更新頻率控制.緩存更新都集中處理,對于發布訂閱模式,同一類主題(Topic)的鍵值集中處理.Cache Updater可以控制對同一鍵值的在短期內的更新頻率(參見下文排重優化).

后臺任務處理

典型的后臺任務處理應用包括工單處理、火車票預訂系統、機票選座等.我們所面對的問題是為運營人員創建工單.一次可以為多個運營人員創建多個工單.這個應用場景和火車票購買非常類似.工單相對來說更加抽象,所以,下文會結合火車票購買和運營人員工單分配這兩種場景同時講解.

典型的工單創建要經歷兩個階段：數據篩選階段、工單創建階段.例如,在火車票預訂場景,數據篩選階段用戶選擇特定時間、特定類型的火車,而在工單創建階段,用戶下單購買火車票.

挑戰

工單創建往往會面臨如下挑戰：

• 數據一致性問題.以火車票預訂為例,用戶篩選火車票和最終購買之間往往有一定的時延,意味著兩個操作之間數據是不一致的.在篩選階段,工程師們需決定是否進行車票鎖定,如果不鎖定,則無法保證出票成功.反之,如果在篩選地時候鎖定車票,則會大大降低系統效率和出票吞吐量.

• 約束問題.工單創建需要滿足很多約束,主要包含兩種類型：動態約束,與操作者的操作行為有關,例如購買幾張火車票的決定往往發生在篩選最后階段.隱性約束,這種約束很難通過界面進行展示,例如一個用戶購買了5張火車票,這些票應該是在同一個車廂的臨近位置.

• 優化問題.工單創建往往是約束下的優化,這是典型的統籌優化問題,而統籌優化往往需要比較長的時間.
• 響應時間問題.對于多任務工單,一個請求意味著多個任務產生.這些任務的創建往往需要遵循事務性原則,即All or Nothing.在數據層面,這意味著工單之間需要滿足串行化需求(Serializability).大數據量的串行化往往意味著鎖沖突延遲甚至失敗.無論是延遲機制所導致的長時延,還是高創建失敗率,都會大大傷害用戶體驗.

構思

如果將用戶篩選的最終規則做為消息存儲下來,并發送給工單創建系統.此時,工單創建系統將具備創建工單所需的全局信息,具備在滿足各種約束的條件下進行統籌優化的能力.如果工單創建階段采用單實例部署,就可以避免數據鎖定問題,同時也意味著沒有鎖沖突,所以也不會有死鎖或任務延遲問題.

居于以上思路,在多工單處理系統的模型中,篩選階段的規則創建系統將充當生產者角色,工單創建系統將充當消費者角色,篩選規則將作為消息在兩者之間進行傳遞.這就是典型的分布式隊列編程架構.根據工單創建量的不同,可以采用數據庫或開源的分布式消息中間件作為分布式隊列.

架構

該架構流程如下圖：

• 用戶首選進行規則創建,這個過程主要是一些搜索篩選操作.

• 用戶點擊工單創建,TicketRule Generator將把所有的篩選性組裝成規則消息并發送到隊列里面去.

• Ticket Generator作為一個消費者,實時從隊列中讀取工單創建請求,開始真正創建工單.

采用該架構,我們在數據鎖定、運籌優化、原子性問題都能得到比較好成果：

• 數據鎖定推遲到工單創建階段,可以減少數據鎖定范圍,最大程度的降低工單創建對其他在線操作的影響范圍.
• 如果需要進行統籌優化,可以將Ticket Generator以單例模式進行部署(參見單例服務優化).這樣,Ticket Generator可以讀取一段時間內的工單請求,進行全局優化.例如,在我們的項目中,在某種條件下,運營人員需要滿足分級公平原則,即相同級別的運營人員的工單數量應該接近,不同級別的運營人員工單數量應該有所區分.如果不集中進行統籌優化,實現這種優化規則將會很困難.
• 保障了約束完整性.例如,在我們的場景里面,每個運營人員每天能夠處理的工單是有數量限制的,如果采用并行處理的方式,這種完整性約束將會很難實施.

3.優化篇

接下來重點闡述工程師運用分布式隊列編程構架的時候,在生產者、分布式隊列以及消費者這三個環節的注意點以及優化建議.

確定采用分布式隊列編程模型之后,主體架構就算完成了,但工程師的工作還遠遠未結束.天下事必做于細,細節是一個不錯的架構向一個優秀的系統進階的關鍵因素.優化篇選取了作者以及其同事在運用分布式隊列編程模型架構時所碰到的典型問題和解決方案.

這里些問題出現的頻率較高,如果你經驗不夠,很可能會“踩坑”.希望通過這些講解,幫助讀者降低分布式隊列編程模型的使用門檻.本文將對分布式隊列編程模型的三種角色：生產者(Producer),分布式隊列(Queue),消費者(Consumer)分別進行優化討論.

生產者優化

在分布式隊列編程中,生產者往往并非真正的生產源頭,只是整個數據流中的一個節點,這種生產者的操作是處理－轉發(Process-Forward)模式.

這種模式給工程師們帶來的第一個問題是吞吐量問題.這種模式下運行的生產者,一邊接收上游的數據,一邊將處理完的數據發送給下游.本質上,它是一個非常經典的數學問題,其抽象模型是一些沒有蓋子的水箱,每個水箱接收來自上一個水箱的水,進行處理之后,再將水發送到下一個水箱.

工程師需要預測水源的流量、每個環節水箱的處理能力、水龍頭的排水速度,最終目的是避免水溢出水箱,或者盡可能地減小溢出事件的概率.實際上流式編程框架以及其開發者花了大量的精力去處理和優化這個問題.下文的緩存優化和批量寫入優化都是針對該問題的解決方案.

第二個需要考慮的問題是持久化.由于各種原因,系統總是會宕機.如果信息比較敏感,例如計費信息、火車票訂單信息等,工程師們需要考慮系統宕機所帶來的損失,找到讓損失最小化的解決方案.持久化優化重點解決這一類問題.

緩存優化

處于“處理－轉發”模式下運行的生產者往往被設計成請求驅動型的服務,即每個請求都會觸發一個處理線程,線程處理完后將結果寫入分布式隊列.如果由于某種原因隊列服務不可用,或者性能惡化,隨著新請求的到來,生產者的處理線程就會產生堆積.這可能會導致如下兩個問題：

• 系統可用性降低.由于每個線程都需要一定的內存開銷,線程過多會使系統內存耗盡,甚至可能產生雪崩效應導致最終完全不可用.
• 信息丟失.為了避免系統崩潰,工程師可能會給請求驅動型服務設置一個處理線程池,設置最大處理線程數量.這是一種典型的降級策略,目的是為了系統崩潰.但是,后續的請求會因為沒有處理線程而被迫阻塞,最終可能產生信息丟失.例如：對于廣告計費采集,如果采集系統因為線程耗盡而不接收客戶端的計費行為,這些計費行為就會丟失.

緩解這類問題的思路來自于CAP理論,即通過降低一致性來提高可用性.生產者接收線程在收到請求之后第一時間不去處理,直接將請求緩存在內存中(犧牲一致性),而在后臺啟動多個處理線程從緩存中讀取請求、進行處理并寫入分布式隊列.

與線程所占用的內存開銷相比,大部分的請求所占內存幾乎可以忽略.通過在接收請求和處理請求之間增加一層內存緩存,可以大大提高系統的處理吞吐量和可擴展性.這個方案本質上是一個內存生產者消費者模型.

批量寫入優化

如果生產者的請求過大,寫分布式隊列可能成為性能瓶頸,有如下幾個因素：

• 隊列自身性能不高.

• 分布式隊列編程模型往往被應用在跨機房的系統里面,跨機房的網絡開銷往往容易成為系統瓶頸.
• 消息確認機制往往會大大降低隊列的吞吐量以及響應時間.

如果在處理請求和寫隊列之間添加一層緩存,消息寫入程序批量將消息寫入隊列,可以大大提高系統的吞吐量.原因如下：

• 批量寫隊列可以大大減少生產者和分布式隊列的交互次數和消息傳輸量.特別是對于高吞吐小載荷的消息實體,批量寫可以顯著降低網絡傳輸量.
• 對于需要確認機制的消息,確認機制往往會大大降低隊列的吞吐量以及響應時間,某些高敏感的消息需要多個消息中間件代理同時確認,這近一步惡化性能.在生產者的應用層將多條消息批量組合成一個消息體,消息中間件就只需要對批量消息進行一次確認,這可能會數量級的提高消息傳輸性能.

持久化優化

通過添加緩存,消費者服務的吞吐量和可用性都得到了提升.但緩存引入了一個新問題——內存數據丟失.對于敏感數據,工程師需要考慮如下兩個潛在問題：

• 如果內存中存在未處理完的請求,而某些原因導致生產者服務宕機,內存數據就會丟失而可能無法恢復.
• 如果分布式隊列長時間不可用,隨著請求數量的不斷增加,最終系統內存可能會耗盡而崩潰,內存的消息也可能丟失.

所以緩存中的數據需要定期被持久化到磁盤等持久層設備中,典型的持久化觸發策略主要有兩種：

• 定期觸發,即每隔一段時間進行一次持久化.

• 定量觸發,即每當緩存中的請求數量達到一定閾值后進行持久化.
• 是否需要持久化優化,以及持久化策略應該由請求數據的敏感度、請求量、持久化性能等因素共同決定.

中間件選型

分布式隊列不等同于各種開源的或者收費的消息中間件,甚至在一些場景下完全不需要使用消息中間件.但是,消息中間件產生的目的就是解決消息傳遞問題,這為分布式隊列編程架構提供了很多的便利.在實際工作中,工程師們應該將成熟的消息中間件作為隊列的首要備選方案.

本節對消息中間件的功能、模型進行闡述,并給出一些消息中間件選型、部署的具體建議.

中間件的功能

明白一個系統的每個具體功能是設計和架構一個系統的基礎.典型的消息中間件主要包含如下幾個功能：

• 消息接收

• 消息分發

• 消息存儲
• 消息讀取

概念模型

抽象的消息中間件模型包含如下幾個角色：

• 發送者和接收者客戶端(Sender/Receiver Client),在具體實施過程中,它們一般以庫的形式嵌入到應用程序代碼中.

• 代理服務器(Broker Server),它們是與客戶端代碼直接交互的服務端代碼.

• 消息交換機(Exchanger),接收到的消息一般需要通過消息交換機(Exchanger)分發到具體的消息隊列中.

• 消息隊列,一般是一塊內存數據結構或持久化數據.

• 概念模型如下圖：

• 為了提高分發性能,很多消息中間件把消息代理服務器的拓撲圖發送到發送者和接收者客戶端(Sender/Receiver Client),如此一來,發送源可以直接進行消息分發.

選型標準

要完整的描述消息中間件各個方面非常困難,大部分良好的消息中間件都有完善的文檔,這些文檔的長度遠遠超過本文的總長度.但如下幾個標準是工程師們在進行消息中間件選型時經常需要考慮和權衡的.

性能

性能主要有兩個方面需要考慮：吞吐量(Throughput)和響應時間(Latency).

不同的消息隊列中間件的吞吐量和響應時間相差甚遠,在選型時可以去網上查看一些性能對比報告.

對于同一種中間件,不同的配置方式也會影響性能.主要有如下幾方面的配置：

• 是否需要確認機制,即寫入隊列后,或從隊列讀取后,是否需要進行確認.確認機制對響應時間的影響往往很大.

• 能否批處理,即消息能否批量讀取或者寫入.批量操作可以大大減少應用程序與消息中間件的交互次數和消息傳遞量,大大提高吞吐量.

• 能否進行分區(Partition).將某一主題消息隊列進行分區,同一主題消息可以有多臺機器并行處理.這不僅僅能影響消息中間件的吞吐量,還決定著消息中間件是否具備良好的可伸縮性(Scalability).
• 是否需要進行持久化.將消息進行持久化往往會同時影響吞吐量和響應時間.

可靠性

可靠性主要包含：可用性、持久化、確認機制等.

高可用性的消息中間件應該具備如下特征：

• 消息中間件代理服務器(Broker)具有主從備份.即當一臺代理服務宕機之后,備用服務器能接管相關的服務.

• 消息中間件中緩存的消息是否有備份、并持久化.
• 根據CAP理論,高可用、高一致性以及網絡分裂不可兼得.根據作者的觀察,大部分的消息中間件在面臨網絡分裂的情況下下,都很難保證數據的一致性以及可用性. 很多消息中間件都會提供一些可配置策略,讓使用者在可用性和一致性之間做權衡.

高可靠的消息中間件應該確保從發送者接收到的消息不會丟失.中間件代理服務器的宕機并不是小概率事件,所以保存在內存中的消息很容易發生丟失.大部分的消息中間件都依賴于消息的持久化去降低消息丟失損失,即將接收到的消息寫入磁盤.即使提供持久化,仍有兩個問題需要考慮：

• 磁盤損壞問題.長時間來看,磁盤出問題的概率仍然存在.

• 性能問題.與操作內存相比,磁盤I/O的操作性能要慢幾個數量級.頻繁持久化不僅會增加響應時間,也會降低吞吐量.
• 解決這兩個問題的一個解決方案就是：多機確認,定期持久化.即消息被緩存在多臺機器的內存中,只有每臺機器都確認收到消息,才跟發送者確認(很多消息中間件都會提供相應的配置選項,讓用戶設置最少需要多少臺機器接收到消息).由于多臺獨立機器同時出故障的概率遵循乘法法則,指數級降低,這會大大提高消息中間件的可靠性.

確認機制本質上是通訊的握手機制(Handshaking).如果沒有該機制,消息在傳輸過程中丟失將不會被發現.高敏感的消息要求選取具備確認機制的消息中間件.當然如果沒有接收到消息中間件確認完成的指令,應用程序需要決定如何處理.典型的做法有兩個：

• 多次重試.
• 暫存到本地磁盤或其它持久化媒介.

客戶端接口所支持語言

采用現存消息中間件就意味著避免重復造輪子.如果某個消息中間件未能提供對應語言的客戶端接口,則意味著極大的成本和兼容性問題.

投遞策略(Delivery policies)

投遞策略指的是一個消息會被發送幾次.主要包含三種策略：最多一次(At most Once )、最少一次(At least Once)、僅有一次(Exactly Once).

在實際應用中,只考慮消息中間件的投遞策略并不能保證業務的投遞策略,因為接收者在確認收到消息和處理完消息并持久化之間存在一個時間窗口.例如,即使消息中間件保證僅有一次(Exactly Once),如果接收者先確認消息,在持久化之前宕機,則該消息并未被處理.

從應用的角度,這就是最多一次(At most Once).反之,接收者先處理消息并完成持久化,但在確認之前宕機,消息就要被再次發送,這就是最少一次(At least Once). 如果消息投遞策略非常重要,應用程序自身也需要仔細設計.

消費者優化

消費者是分布式隊列編程中真正的數據處理方,數據處理方最常見的挑戰包括：有序性、串行化(Serializability)、頻次控制、完整性和一致性等.

挑戰

有序性

在很多場景下,如何保證隊列信息的有序處理是一個棘手的問題.如下圖,假定分布式隊列保證請求嚴格有序,請求ri2和ri1都是針對同一數據記錄的不同狀態,ri2的狀態比ri1的狀態新.T1、T2、T3和T4代表各個操作發生的時間,并且 T1 < T2 < T3 < T4(”<“代表早于).

采用多消費者架構,這兩條記錄被兩個消費者(Consumer1和Consumer2)處理后更新到數據庫里面.Consumer1雖然先讀取ri1但是卻后寫入數據庫,這就導致,新的狀態被老的狀態覆蓋,所以多消費者不保證數據的有序性.

串行化

很多場景下,串行化是數據處理的一個基本需求,這是保證數據完整性、可恢復性、事務原子性等的基礎.為了在并行計算系統里實現串行化,一系列的相關理論和實踐算法被提出.對于分布式隊列編程架構,要在在多臺消費者實現串行化非常復雜,無異于重復造輪子.

頻次控制

有時候,消費者的消費頻次需要被控制,可能的原因包括：

• 費用問題.如果每次消費所引起的操作都需要收費,而同一個請求消息在隊列中保存多份,不進行頻次控制,就會導致無謂的浪費.
• 性能問題.每次消費可能會引起對其他服務的調用,被調用服務希望對調用量有所控制,對同一個請求消息的多次訪問就需要有所控制.

完整性和一致性

完整性和一致性是所有多線程和多進程的代碼都面臨的問題.在多線程或者多進程的系統中考慮完整性和一致性往往會大大地增加代碼的復雜度和系統出錯的概率.

單例服務優化

幾乎所有串行化理論真正解決的問題只有一個：性能. 所以,在性能允許的前提下,對于消費者角色,建議采用單實例部署.通過單實例部署,有序性、串行化、完整性和一致性問題自動獲得了解決.另外,單實例部署的消費者擁有全部所需信息,它可以在頻次控制上采取很多優化策略.

天下沒有免費的午餐.同樣,單實例部署并非沒有代價,它意味著系統可用性的降低,很多時候,這是無法接受的.解決可用性問題的最直接的思路就是冗余(Redundancy).最常用的冗余方案是Master-slave架構,不過大部分的Master-slave架構都是Active/active模式,即主從服務器都提供服務.

例如,數據庫的Master-slave架構就是主從服務器都提供讀服務,只有主服務器提供寫服務.大部分基于負載均衡設計的Master-slave集群中,主服務器和從服務器同時提供相同的服務.這顯然不滿足單例服務優化需求.

有序性和串行化需要Active/passive架構,即在某一時刻只有主實例提供服務,其他的從服務等待主實例失效.這是典型的領導人選舉架構,即只有獲得領導權的實例才能充當實際消費者,其他實例都在等待下一次選舉.采用領導人選舉的Active/passive架構可以大大緩解純粹的單實例部署所帶來的可用性問題.

令人遺憾的是,除非工程師們自己在消費者實例里面實現Paxos等算法,并在每次消息處理之前都執行領導人選舉.否則,理論上講,沒有方法可以保障在同一個時刻只有一個領導者.而對每個消息都執行一次領導人選舉,顯然性能不可行.

實際工作中,最容易出現的問題時機發生在領導人交接過程中,即前任領導人實例變成輔助實例,新部署實例開始承擔領導人角色.為了平穩過渡,這兩者之間需要有一定的通訊機制,但是,無論是網絡分區(Network partition)還是原領導人服務崩潰都會使這種通訊機制變的不可能.

對于完整性和一致性要求很高的系統,我們需要在選舉制度和交接制度這兩塊進行優化.

領導人選舉架構

典型的領導人選舉算法有Paxos、ZAB( ZooKeeper Atomic Broadcast protocol).為了避免重復造輪子,建議采用ZooKeeper的分布式鎖來實現領導人選舉.典型的ZooKeeper實現算法如下：

Let ELECTION be a path of choice of the application. To volunteer to be a leader:

1.Create znode z with path “ELECTION/guid-n_” with both SEQUENCE and EPHEMERAL flags;

2.Let C be the children of “ELECTION”, and i be the sequence number of z;

3.Watch for changes on “ELECTION/guid-n_j”, where j is the largest sequence number such that j < i and n_j is a znode in C;

Upon receiving a notification of znode deletion:

1.Let C be the new set of children of ELECTION;

2.If z is the smallest node in C, then execute leader procedure;

3.Otherwise, watch for changes on “ELECTION/guid-n_j”, where j is the largest sequence number such that j < i and n_j is a znode in C;

領導人交接架構

領導人選舉的整個過程發生在ZooKeeper集群中,各個消費者實例在這場選舉中只充當被告知者角色(Learner).領導人選舉算法,只能保證最終只有一個Leader被選舉出來,并不保障被告知者對Leader的理解是完全一致的.

本質上,上文的架構里,選舉的結果是作為令牌(Token)傳遞給消費者實例,消費者將自身的ID與令牌進行對比,如果相等,則開始執行消費操作.所以當發生領導人換屆的情況,不同的Learner獲知新Leader的時間并不同.

例如,前任Leader如果因為網絡問題與ZooKeeper集群斷開,前任Leader只能在超時后才能判斷自己是否不再承擔Leader角色了,而新的Leader可能在這之前已經產生.另一方面,即使前任Leader和新Leader同時接收到新Leader選舉結果,某些業務的完整性要求迫使前任Leader仍然完成當前未完成的工作.

以上的講解非常抽象,生活中卻給了一些更加具體的例子.眾所周知,美國總統候選人在選舉結束后并不直接擔任美國總統,從選舉到最終承擔總統角色需要一個過渡期.對于新當選Leader的候選人而言,過渡期間稱之為加冕階段(Inauguration).對于即將卸任的Leader,過渡期稱為交接階段(HandOver).

所以一個基于領導人選舉的消費者從加冕到卸任經歷三個階段：Inauguration、Execution、HandOver.在加冕階段,新領導需要進行一些初始化操作.Execution階段是真正的隊列消息處理階段.在交接階段,前任領導需要進行一些清理操作.

類似的,為了解決領導人交接問題,所有的消費者從代碼實現的角度都需要實現類似ILeaderCareer接口.這個接口包含三個方發inaugurate(),handOver()和execute().某個部署實例(Learner)在得知自己承擔領導人角色后,需要調用inaugurate()方法,進行加冕.主要的消費邏輯通過不停的執行execute()實現,當確認自己不再承擔領導人之后,執行handOver()進行交接.

如果承擔領導人角色的消費者,在執行execute()階段得知自己將要下臺,根據消息處理的原子性,該領導人可以決定是否提前終止操作.如果整個消息處理是一個原子性事務,直接終止該操作可以快速實現領導人換屆.否則,前任領導必須完成當前消息處理后,才進入交接階段.這意味著新的領導人,在inaugurate()階段需要進行一定時間的等待.

排重優化

頻次控制是一個經典問題.對于分布式隊列編程架構,相同請求重復出現在隊列的情況并不少見.如果相同請求在隊列中重復太多,排重優化就顯得很必要.分布式緩存更新是一個典型例子,所有請求都被發送到隊列中用于緩存更新.如果請求符合典型的高斯分布,在一段時間內會出現大量重復的請求,而同時多線程更新同一請求緩存顯然沒有太大的意義.

排重優化是一個算法,其本質是基于狀態機的編程,整個講解通過模型、構思和實施三個步驟完成.

模型

進行排重優化的前提是大量重復的請求.在模型這一小節,我們首先闡述重復度模型、以及不同重復度所導致的消費模型,最后基于這兩個模型去講解排重狀態機.

重復度模型

首先我們給出最小重復長度的概念.同一請求最小重復長度：同一請求在隊列中的重復出現的最小間距.例如,請求ri第一次出現在位置3,第二次出現在10,最小重復長度等于7.

是否需要進行排重優化取決于隊列中請求的重復度.由于不同請求之間并不存在重復的問題,不失一般性,這里的模型只考了單個請求的重復度,重復度分為三個類：無重復、稀疏重復、高重復.

• 無重復：在整個請求過程,沒有任何一個請求出現一次以上.

• 稀疏重復：主要的請求最小重復長度大于消費隊列長度.
• 高重復：大量請求最小重復長度小于消費隊列長度.

對于不同的重復度,會有不同的消費模型.

無重復消費模型

在整個隊列處理過程中,所有的請求都不相同,如下圖：

稀疏重復消費模型

當同一請求最小重復長度大于消費者隊列長度,如下圖.假定有3個消費者,Consumer1將會處理r1,Consumer2將會處理r2,Consumer3將會處理r3,如果每個請求處理的時間嚴格相等,Consumer1在處理完r1之后,接著處理r4,Consumer2將會處理r2之后會處理r1.雖然r1被再次處理,但是任何時刻,只有這一個消費者在處理r1,不會出現多個消費者同時處理同一請求的場景.

高重復消費模型

如下圖,仍然假定有3個消費者,隊列中前面4個請求都是r1,它會同時被3個消費者線程處理：

顯然,對于無重復和稀疏重復的分布式隊列,排重優化并不會帶來額外的好處.排重優化所針對的對象是高重復消費模型,特別是對于并行處理消費者比較多的情況,重復處理同一請求,資源消耗極大.

排重狀態機

排重優化的主要對象是高重復的隊列,多個消費者線程或進程同時處理同一個冪等請求只會浪費計算資源并延遲其他待請求處理.所以,排重狀態機的一個目標是處理唯一性,即：同一時刻,同一個請求只有一個消費者處理.

如果消費者獲取一條請求消息,但發現其他消費者正在處理該消息,則當前消費者應該處于等待狀態.如果對同一請求,有一個消費者在處理,一個消費者在等待,而同一請求再次被消費者讀取,再次等待則沒有意義.

所以,狀態機的第二個目標是等待唯一性,即：同一時刻,同一個請求最多只有一個消費者處于等待狀態.總上述,狀態機的目標是：處理唯一性和等待唯一性.我們把正在處理的請求稱為頭部請求,正在等待的請求稱為尾部請求.

由于狀態機的處理單元是請求,所以需要針對每一個請求建立一個排重狀態機.基于以上要求,我們設計的排重狀態機包含4個狀態Init,Process,Block,Decline.各個狀態之間轉化過程如下圖：

1. 狀態機創建時處于Init狀態.
2. 對Init狀態進行Enqueue操作,即接收一個請求,開始處理(稱為頭部請求),狀態機進入Process狀態.
3. 狀態機處于Process狀態,表明當前有消費者正在處理頭部請求.此時,如果進行Dequeue操作,即頭部請求處理完成,返回Init狀態.如果進行Enqueue操作,即另一個消費者準備處理同一個請求,狀態機進入Block狀態(該請求稱為尾部請求).
4. 狀態機處于Block狀態,表明頭部請求正在處理,尾部請求處于阻塞狀態.此時,進行Dequeue操作,即頭部請求處理完成,返回Process狀態,并且尾部請求變成頭部請求,原尾部請求消費者結束阻塞狀態,開始處理.進行Enqueue操作,表明一個新的消費者準備處理同一個請求,狀態機進入Decline狀態.
5. 狀態機進入Decline狀態,根據等待唯一性目標,處理最新請求的消費者將被拋棄該消息,狀態機自動轉換回Block狀態.

構思

狀態機描述的是針對單個請求操作所引起狀態變化,排重優化需要解決隊列中所有請求的排重問題,需要對所有請求的狀態機進行管理.這里只考慮單虛擬機內部對所有請求狀態機的管理,對于跨虛擬機的管理可以采用類似的方法.對于多狀態機管理主要包含三個方面：一致性問題、完整性問題和請求緩存驅逐問題.

一致性問題

一致性在這里要求同一請求的不同消費者只會操作一個狀態機.由于每個請求都產生一個狀態機,系統將會包含大量的狀態機.為了兼顧性能和一致性,我們采用ConcurrentHashMap保存所有的狀態機.用ConcurrentHashMap而不是對整個狀態機隊列進行加鎖,可以提高并行處理能力,使得系統可以同時操作不同狀態機.

為了避免處理同一請求的多消費者線程同時對ConcurrentHashMap進行插入所導致狀態機不一致問題,我們利用了ConcurrentHashMap的putIfAbsent()方法.代碼方案如下,key2Status用于存儲所有的狀態機.

消費者在處理請求之前,從狀態機隊列中讀取排重狀態機TrafficAutomate.如果沒有找到,則創建一個新的狀態機,并通過putIfAbsent()方法插入到狀態機隊列中.

完整性問題

完整性要求保障狀態機Init,Process,Block,Decline四種狀態正確、狀態之間的轉換也正確.由于狀態機的操作非常輕量級,兼顧完整性和降低代碼復雜度,我們對狀態機的所有方法進行加鎖.

請求緩存驅逐問題(Cache Eviction)

如果不同請求的數量太多,內存永久保存所有請求的狀態機的內存開銷太大.所以,某些狀態機需要在恰當的時候被驅逐出內存.這里有兩個思路：

• 當狀態機返回Init狀態時,清除出隊列.
• 啟動一個后臺線程,定時掃描狀態機隊列,采用LRU等標準緩存清除機制.

標識問題

每個請求對應于一個狀態機,不同的狀態機采用不同的請求進行識別.

對于同一狀態機的不同消費者,在單虛擬機方案中,我們采用線程id進行標識.

實施

排重優化的主要功能都是通過排重狀態機(TrafficAutomate)和狀態機隊列(QueueCoordinator)來實施的.排重狀態機描述的是針對單個請求的排重問題,狀態機隊列解決所有請求狀態機的排重問題.

狀態機實施(TrafficAutomate)

根據狀態機模型,其主要操作為enQueue和deQueue,其狀態由頭部請求和尾部請求的狀態共同決定,所以需要定義兩個變量為head和tail,用于表示頭部請求和尾部請求.為了確保多線程操作下狀態機的完整性(Integraty),所有的操作都將加上鎖.

enQueue操作

當一個消費者執行enQueue操作時：如果此時尾部請求不為空,根據等待唯一性要求,返回DECLINE,當前消費者應該拋棄該請求；如果頭部請求為空,返回ACCPET,當前消費者應該立刻處理該消息；否則,返回BLOCK,該消費者應該等待,并不停的查看狀態機的狀態,一直到頭部請求處理完成.enQueue代碼如下：

deQueue操作

對于deQueue操作,首先將尾部請求賦值給頭部請求,并將尾部請求置為無效.deQueue代碼如下：

狀態機隊列實施(QueueCoordinator)

接口定義

狀態機隊列集中管理所有請求的排重狀態機,所以其操作和單個狀態機一樣,即enQueue和deQueuqe接口.這兩個接口的實現需要識別特定請求的狀態機,所以它們的入參應該是請求.為了兼容不同類型的請求消息,我們采用了Java泛型編程.接口定義如下：

enQueue操作

enQueue操作過程如下：

首先,根據傳入的請求key值,獲取狀態機, 如果不存在則創建一個新的狀態機,并保存在ConcurrentHashMap中.

接下來,獲取線程id作為該消費者的唯一標識,并對對應狀態機進行enQueue操作.

如果狀態機返回值為ACCEPT或者DECLINE,返回業務層處理代碼,ACCEPT意味著業務層需要處理該消息,DECLINE表示業務層可以拋棄當前消息.如果狀態機返回值為Block,則該線程保持等待狀態.

在某些情況下,頭部請求線程可能由于異常,未能對狀態機進行deQueue操作(作為組件提供方,不能假定所有的規范被使用方實施).為了避免處于阻塞狀態的消費者無期限地等待,建議對狀態機設置安全超時時限.超過了一定時間后,狀態機強制清空頭部請求,返回到業務層,業務層開始處理該請求.

代碼如下：

deQueue操作

deQueue操作首先從ConcurrentHashMap獲取改請求所對應的狀態機,接著獲取該線程的線程id,對狀態機進行deQueue操作.

enQueue代碼如下：

源代碼

完整源代碼可以在QueueCoordinator獲取.鏈接：

https://github.com/dinglau2008/QueueCoordinator/tree/master/src

文章出處：美團點評技術團隊

轉載請注明本頁網址：
http://www.snjht.com/jiaocheng/4455.html