消息隊列作為智聯(lián)招聘非常重要的平臺級服務(wù)負責全業(yè)務(wù)線的消息投遞。有很多非常典型的業(yè)務(wù)場景，我們用一個業(yè)務(wù)場景簡歷投遞來說明消息隊列為業(yè)務(wù)提供的支持

 

 

圖 1. 簡歷投遞業(yè)務(wù)

當 C 端用戶發(fā)生一次簡歷投遞的時候會先發(fā)送一條消息到消息隊列服務(wù)，C 端中臺、B 端中臺以及平臺級的基礎(chǔ)服務(wù)會從消息隊列消費這條消息進行自己的業(yè)務(wù)邏輯處理比如寫 DB、通知 B 端企業(yè)等，在這個場景中消息隊列為投遞業(yè)務(wù)提供了很好異步解耦支持，在高峰期的時候可以提供很好的削峰作用以保障各業(yè)務(wù)系統(tǒng)的穩(wěn)定運行。

上面的這個場景是非常典型的工作隊列場景，在工作隊列中大多數(shù)是為了實現(xiàn)業(yè)務(wù)場景支持在線服務(wù)而設(shè)定的，往往具有以下特點：

♦ 一條消息會被多個業(yè)務(wù)方消費
♦ 多個業(yè)務(wù)方之間廣播方式消費 (每個業(yè)務(wù)方消費完整的一份數(shù)據(jù))
♦ 單個業(yè)務(wù)方采用集群消費模式 (每個 consumer 消費部分數(shù)據(jù))
♦ 每條消息都需要確保送達，消息隊列會采用重試的機制來保證這一點
♦ 重要業(yè)務(wù)消息需要提供跟蹤機制可以查詢整個消息的生命周期
♦ 還有一些業(yè)務(wù)的需求會使用到延時消息，定時消息等。

基于 RabbitMQ 的自研 MQService

RabbitMQ 作為一款非常成熟的消息隊列產(chǎn)品可以很好的應對工作隊列的場景，當然也有一些不足比如單隊列的擴展能力、延時消息支持的不夠好等。我們在 RabbitMQ 基礎(chǔ)上又做了一層抽象 (MQService)，將 RabbitMQ 看做一個消息服務(wù)的存儲節(jié)點來用，在 Zookeeper 中會記錄 Topic 的數(shù)據(jù)在 RabbitMQ 節(jié)點的分布并增加了容錯的特性來保證存儲節(jié)點失敗的情況下可以持續(xù)提供消息寫入能力。在招聘旺季消息隊列每天約承載數(shù) 10 億的消息投遞。

MQService 整體結(jié)構(gòu)如下：

 

 

圖 2.MQService 架構(gòu)

用戶可以通過 Thrift 協(xié)議、Http 協(xié)議、MQTT 來做消息的發(fā)送和消費。用戶也可以注冊 Http 回調(diào)接口來消費消息。默認的 Java 客戶端封裝了 Thrift 協(xié)議及 MQTT 的消息生產(chǎn)及消費，其他語言并沒有封裝對應的客戶端而是通過 Http 協(xié)議進行消息的生產(chǎn)和消費，整體智聯(lián)招聘也是以 Java 做后臺服務(wù)為主。

接下來我們看一下通過 zookeeper 維護的 Topic 與 RabbitMQ 分組以及 RabbitMQ 節(jié)點的關(guān)系。

 

 

每個 Topic 都對對應到一個 Group，每個 Group 下會掛一些 RabbitMQ 節(jié)點。當 producer 發(fā)送一條消息時 MQService 會從緩存中拿到對應這個 Topic 可用的 RabbitMQ 節(jié)點的列表，MQService 會通過負載均衡策略選擇其中的一個 RabbitMQ 節(jié)點進行寫入，寫入失敗會重試下一個節(jié)點直到寫入成功。單個節(jié)點如果寫入失敗次數(shù)在一定時間內(nèi)達到一個特定值會觸發(fā)熔斷機制，單個 RabbitMQ 節(jié)點在熔斷期間不對外提供寫入及查詢服務(wù)。

通過上面的介紹，大家應該可以對 MQService 有一個直觀的印象，這里不再詳細展開來介紹實現(xiàn)的細節(jié)。

以 Kafka 為中心的流批處理

首先 Kafka 在智聯(lián)有招聘有大規(guī)模的應用，每天的數(shù)據(jù)傳輸量大約在數(shù)十 TB 量級，覆蓋的范圍包括 ELK、實時計算等。我們還是以計算每天不同時間端的投遞量為例子來介紹 Kakfa 在這個場景下的使用。

 

 

圖 3. 基于 Kafka 的 Streaming 模型

這是一個非常經(jīng)典的流式計算的架構(gòu)，通過采集業(yè)務(wù)日志入 Kafka，再通過 Spark/Flink 之類的計算框架做計算工作。在計算投遞量的場景中，我們通過將計算結(jié)果按小時以及職位為維度將計算結(jié)果保存在 MySQL 中，也會將明細數(shù)據(jù)存儲在 Hive 中供離線計算使用。

那么很容易發(fā)現(xiàn)同一個業(yè)務(wù)場景但是數(shù)據(jù)來源是不一樣的，一個是業(yè)務(wù)方發(fā)送至 MQService 的一個是通過 Logstash 采集業(yè)務(wù)端日志來做的，那么我們?nèi)绾蝸肀ＷC這兩份數(shù)據(jù)的一致性?即使不通過采集日志，業(yè)務(wù)方去雙寫又如何來保證一致性呢?這樣也給用戶帶來額外的負擔。

矛盾點在于 MQService 很難融入到流行的實時計算框架或者批處理框架，而 Kafka 也應用在工作隊列模式下顯得有點力不從心。主要體現(xiàn)在 Kakfa 數(shù)據(jù)消費的支持，Partition 數(shù)量與 Consumer 數(shù)量的綁定造成的 Partition 數(shù)量要跟著 consumer 的消費能力決定，業(yè)務(wù)方處理數(shù)據(jù)很難保證一批數(shù)據(jù)都能夠成功處理，做 offset commit 的時候也是無法達到用戶的預期。這是基于產(chǎn)品在業(yè)務(wù)場景的匹配上而討論論的，就像 Kafka 介紹的那樣 (A distributed streaming platform)。

因此在 2018 年初時我們提出了通過一套方案來解決“工作隊列 + Streaming”的想法，也就是事件中心，期望事件中心可以承載智聯(lián)全業(yè)務(wù)線用戶行為、中臺以及后臺的業(yè)務(wù)事件傳遞。產(chǎn)品和業(yè)務(wù)系統(tǒng)在服務(wù)過程中會產(chǎn)生事件，事件是在先前定義好的，對于事件的生產(chǎn)方無需關(guān)心事件的消費，只需要關(guān)心事件的格式定義以及事件的生產(chǎn)。事件的消費方可以在事件中心去查閱自己想要的訂閱的事件來申請訂閱，甚至是數(shù)據(jù)產(chǎn)品也可以在事件中心去找找靈感。

這樣我們不需要關(guān)心兩個消息中間件數(shù)據(jù)的一致性問題，一份數(shù)據(jù)就可以匹配“工作隊列 + Streaming”場景，對資源消耗、系統(tǒng)運維都有很好的改善。

工作隊列 + Streaming 場景全新訴求

有了這個想法之后我們開始總結(jié)我們需要的是一個什么樣的產(chǎn)品，然后圍繞我們的需求去做設(shè)計工作和技術(shù)調(diào)研工作。我們總結(jié)了我們一些訴求如下：

容災能力及一致性

數(shù)據(jù)做分布式存儲并且在分布式環(huán)境中要保證一致性。有一些重要的業(yè)務(wù)是依賴消息可靠性以及數(shù)據(jù)一致性的，所以在技術(shù)選型的時候如果在一個支持一致性的模型下去弱化一致性提升可用性是比較容易的，但是如果在一個沒有一致性模型的方案上去做一致性這將會需要一個很大的改動。

單 Topic 擴展能力

就像在 MQService 描述的那樣，同一個 Topic 可以利用多個節(jié)點來做橫向擴展。Kafka 在這一點做了很好的抽象 (Partition)。同一個 Partition 的事件可以提供順序性消費。

累計簽收與單條簽收

累計簽收主要應用在 Streaming 場景下，而單條簽收可以很好的匹配工作隊列的場景，就像一次簡歷投遞的業(yè)務(wù)處理，業(yè)務(wù)本身沒有順序性的要求，單條簽收可以很好的支持消費者的消費能力擴展。而在累計簽收模式下單分區(qū)是要保證順序性的。

事件回溯能力

我們需要根據(jù)不同的事件來決定保留的時長或大小，可以為一些想要拿到歷史事件的業(yè)務(wù)提供支持，我們也可以看到這也是 MQService(上文提到的) 薄弱的地方，MQService 是無法給用戶提供回溯能力的。

基于上面的一些主要的特性我們開始了技術(shù)選型的調(diào)研工作。經(jīng)過了一段時間的調(diào)研工作后，我們發(fā)現(xiàn)開源的消息中間件產(chǎn)品兼顧容災能力和一致性的產(chǎn)品幾乎沒有，因此我們產(chǎn)生了一個想法，那就是基于一個強一致性的分布式日志存儲系統(tǒng)來做隊列功能的開發(fā)，期間也考慮過使用 Raft 協(xié)議 +Log 存儲，但是最終還是 Bookkeeper 吸引了我們的關(guān)注。Bookeeper 提供了開箱即用的 API，同時它在 Twitter、Hadoop 已經(jīng)有大規(guī)模應用的場景，其穩(wěn)定性以及成熟度等都是可以保證的。因此我們在大約 5 月份的時候已經(jīng)開始做基于 Bookkeeper 事件中心的一些設(shè)計工作，在接觸 Bookkeeper 社區(qū)的的時候才了解到 Apache Puslar，通過了一段時間對 Apache Pulsar 的了解以及對社區(qū)活躍度的觀察，在 Apache Pulsar 社區(qū)小伙伴們的大力支持下，我們決定基于 Apache Pulsar 來搭建我們的事件中心。

為什么選擇 Apache Pulsar ?

Apache Pulsar 有很多特性在滿足事件中心需求的前提了還給了我們更多的驚喜，為更多的場景提供非常好的解決方案。

靈活的可用性和一致性選擇

在每個 Topic 中由一系列的 Ledger 構(gòu)成，每個 Ledger 有三個關(guān)鍵配置：

♦ Ensemble Size (E)
♦ Write Quorum Size (Qw)
♦ Ack Quorum Size (Qa)

Ensemble Size (E) 決定了 Pulsar 寫入 Ledger 可用的 Bookies 池的大小。

Write Quorum (Qw) 是 Pulsar 將要寫入的實際的 Bookies 數(shù)量�？梢缘扔诨蛘咝∮� E。

 

 

圖 4.E = 3 Qw = 3

當 Qw 小于 E 時，以條帶化的方式分配讀 / 寫即每個 Bookie 只提供讀寫請求的子集。因此可以提升吞吐量，降低延遲。這也是提升單個 Partition 吞吐能力的一個很好的方案，這也得益于基于 Segment 為物理單元的存儲設(shè)計。消息通過 Robin 的方式寫入指定的 Bookie，在查詢消息是可以根據(jù) MessageId 取模即能獲得所在的 Bookie 列表。

 

 

圖 5.E = 5 Qw = 3

Ack Quorum (Qa) 是確認寫入 Bookies 的數(shù)量，Pulsar Broker 將確認發(fā)送給客戶端。為了一致性，Qa 應該是：(Qw + 1) / 2 或者更大。

這個特性可以很好的讓我們在可用性和一致性上去做選擇。

訂閱的抽象

單隊列的擴展 Kafka 為我們做了很好的抽象，Apache Pulsar 也基本采用相同的思路。而對于訂閱的抽象，我們認為 Apache Pulsar 再一次為我們做了很好的抽象，通過 3 種不同的訂閱方式來匹配不同的使用場景。

 

 

圖 6.Apache Pulsar 對訂閱的抽象

消息存儲在 Topic 中。邏輯上一個 Topic 是日志結(jié)構(gòu)，每個消息都在這個日志結(jié)構(gòu)中有一個偏移量。Apache Pulsar 使用游標來跟蹤偏移量。生產(chǎn)者將消息發(fā)送到一個指定的 Topic，Apache Pulsar 保證消息一旦被確認就不會丟失 (正確的配置和非整個集群故障的情況下)。

消費者通過訂閱來消費 Topic 中的消息。訂閱是游標 (跟蹤偏移量) 的邏輯實體，并且還根據(jù)不同的訂閱類型提供一些額外的保證

♦ Exclusive(獨享) - 一個訂閱只能有一個消息者消費消息

♦ Shared(共享) - 一個訂閱中同時可以有多個消費者，多個消費者共享 Topic 中的消息

♦ Fail-Over(災備) - 一個訂閱同時只有一個消費者，可以有多個備份消費者。一旦主消費者故障則備份消費者接管。不會出現(xiàn)同時有兩個活躍的消費者。

一個 Topic 可以添加多個訂閱。訂閱不包含消息的數(shù)據(jù)，只包含元數(shù)據(jù)和游標。

Apache Pulsar 通過允許消費者將 Topic 看做在消費者消費確認后刪除消息的隊列，或者消費者可以根據(jù)游標的回放來提供隊列和日志的語義。在底層都使用日志作為存儲模型。

這為我們通過一套系統(tǒng)支持工作隊列和 Streaming 的訴求提供了很好的支持，在工作隊列場景我們使用 share 模式，在 Streaming 模式我們使用 Failover 或者 Exclusive。我們只需要一份數(shù)據(jù)就可以同時支持兩種場景。

更好的 IO 和存儲設(shè)計

當在 Bookie 上寫入數(shù)據(jù)時，首先將該消息寫入日志文件，這是一個預寫日志 (WAL), 它可以幫助 Bookkeeper 在發(fā)生故障時避免數(shù)據(jù)丟失。它與關(guān)系型數(shù)據(jù)庫持久化保證的機制相同。

寫入預寫日志的操作完成后會將數(shù)據(jù)放入緩存。寫入的緩存會在內(nèi)存中做積累并定期進行排序和刷盤。對寫入進行排序以便將同一 Ledger 的條目放在一起，從而提高讀取性能。如果條目以嚴格的時間順序?qū)懭�，在讀取時無法利用磁盤的高效順序操作

Bookkeeper 容許將磁盤 IO 做讀寫分離。寫入都按順序?qū)懭肴罩疚募�可以存儲在專用的磁盤上，并且可以批量刷盤以獲得搞得吞吐量。除此之外從寫入操作來看沒有其他的同步磁盤 IO 操作，數(shù)據(jù)都是寫入到內(nèi)存的緩存區(qū)。

寫緩存通過異步的方式批量將條目寫入到日志文件和 RocksDB，因此，一個磁盤用于同步寫入日志文件，另一個磁盤用于異步寫入數(shù)據(jù)和讀取操作，

 

 

圖 7.Apache Pulsar 的 IO 及存儲設(shè)計

在存儲設(shè)計上 Bookkeeper 以 Segment 為中心設(shè)計對系統(tǒng)擴容、冷熱數(shù)據(jù)分離提供了很好的支持。在擴容方面通過增加 Bookie 節(jié)點就可以分擔整個集群的存儲壓力，在冷熱數(shù)據(jù)分離方面通過將 Segment 搬遷至二級存儲如 S3、OSS 等更廉價的存儲設(shè)備中，支持在線業(yè)務(wù)往往使用 SSD 來做存儲。因此我們可以兼顧熱數(shù)據(jù)的高性能與冷數(shù)據(jù)的大空間存儲。

 

 

圖 8.Bookie 擴容

 

 

圖 9. 冷數(shù)據(jù)搬遷

在 IO 和存儲設(shè)計上以及 Offload 的特性給了我們更多的驚喜，可以更好的為我們在不影響在線業(yè)務(wù)的支持上兼顧大量事件存儲需求的痛點，大大的降低了冷數(shù)據(jù)的存儲成本。我們計劃將冷數(shù)據(jù)存儲至 OSS。

上面挑選了 Apache Pulsar 非常核心的 3 個 messaging 特性來做介紹，這與事件中心的初衷是非常匹配的，然而 Apache Pulsar 遠不止這些，有完善的多租戶特性提供 Topic 的分層次管理，多種 Schema 的支持為數(shù)據(jù)校驗、序列化提供更便捷的方式，輕量級的 Pulsar Function 以及 Pulsar SQL 都是非常值得去探索的特性，這里就不一一展開介紹了。

Apache Pulsar 在智聯(lián)招聘的落地實踐

下面將介紹 Apache Pulsar 在智聯(lián)招聘落地過程中的一些實踐

為 Namespace 設(shè)置合理的 Backlog Quota

Pulsar 為我們提供了 Backlog 的機制能夠記錄每個 Subscription 的消費狀況。也提供了 Backlog Quota 的設(shè)置，主要可以設(shè)置 Backlog 大小以及達到閾值時的控制策略。

Backlog Quota 的控制策略有 3 種：

♦ producer_request_hold
♦ producer_exception
♦ consumer_backlog_eviction

producer_request_hold 作為默認配置，在達到 Backlog 設(shè)置的大小閾值后會 block producer 發(fā)消息操作，這個配置不適合用于消息發(fā)送方是在線業(yè)務(wù)的使用場景。

producer_exception 在達到 Backlog 設(shè)置的大小閾值后，producer 會快速失敗。

consumer_backlog_eviction 在達到 Backlog 設(shè)置的大小閾值后會將 subscription 未簽收的頭部數(shù)據(jù)逐出，可以理解為自動簽收。其實這個和 producer_exception 的區(qū)別在于 producer_exception 對于訂閱方將會丟失尾部數(shù)據(jù)，而 consumer_backlog_eviction 是丟失頭部數(shù)據(jù)。

我們大部分使用 consumer_backlog_eviction 策略。目前 Pulsar 支持在 namespace 級別設(shè)置這個策略，在 2.2.0 版本可以在 broker.conf 文件修改全局策略。將 Backlog 作為一個重點的監(jiān)控項監(jiān)控起來也是非常有必要的，后面會說到這部分。

增加 MaxProducersPerTopic 限制

防止錯誤或者惡意的 Client 使用造成 Broker 維持大量的 Producer 對象。Broker 默認的配置是不限制的，增加限制可以提升 Pulsar Cluster 的安全性。

Pulsar 提供兩種方式來設(shè)定 MaxProducerPerTopic
broker.conf 中設(shè)置 maxProducersPerTopic
通過./pulsar-admin namespaces set-max-producers-per-topic -p

目前我們在 broker.conf 中的 maxProducersPerTopic = 10000，如果 namespace 有個性需求的話通過 ./pulsar-admin namespaces set-max-producers-per-topic -p 設(shè)置。

Apache Pulsar 監(jiān)控與報警

Pulsar 提供豐富的 Prometheus 指標信息輸出，我們可以這些指標信息來做好 Pulsar 的監(jiān)控報警。Pulsar 的客戶端也記錄了豐富的指標，我們做了一個 client 的擴展包將 client 的節(jié)點信息記錄在 Zookeeper 中，由 Prometheus 自動發(fā)現(xiàn)，這樣 Client 端的指標信息由 Prometheus 采集。

配合 Grafana 的監(jiān)控展示，實時了解集群的狀態(tài)

 

 

圖 10. 集群狀態(tài)看板 -1

 

 

圖 11. 集群狀態(tài)看板 -2

 

 

圖 12. 分 Namespace 狀態(tài)展示

報警規(guī)則配置：

♦ Client 發(fā)送失敗次數(shù)
♦ Backlog 超閾值
♦ Rates/Network 超閾值
♦ Client > 50ms 延遲
♦ Broker > 50ms 延遲
♦ Storage Size 超閾值

Pulsar 多集群流量切換

為了避免集群整體不可用，我們通過 Zookeeper 控制 Client 的連接串�；�于 Pulsar Client 的 Service URL Provider 基礎(chǔ)上做的二次開發(fā)。在 Zookeeper 中存儲的 Pulsar 連接串改變的時候，Client 會自動斷掉當前的連接并重新與新的 Pulsar 地址進行連接。

為重要業(yè)務(wù)提供消息鏈路追蹤

我們基于 Pulsar Client 的 Interceptor 接口以及 Zipkin 進行二次開發(fā)，為了實現(xiàn)消息的鏈路跟蹤。消息鏈路跟蹤的方案是通過日志采集統(tǒng)一入 Hbase。每條消息都具備消息鏈路跟蹤成本是昂貴的，并不適用于所有的場景，更適應與一些比較重要切消息量不太大的場景，當然這個根據(jù)不同的業(yè)務(wù)而定。

總結(jié)

我們通過介紹之前的消息中間件在智聯(lián)招聘的應用情況來說明我們的痛點所在，我們計劃打造一個可以解決當下痛點的產(chǎn)品來支撐智聯(lián)招聘的業(yè)務(wù)。我們通過一段時間的技術(shù)選型工作后最終選擇了 Apache Pulsar 作為我們的搭建企業(yè)級事件中心的基礎(chǔ)。

截止目前事件中心接入的事件種類約 100 個，每天產(chǎn)生 5 億事件量。部分業(yè)務(wù)通過灰度的方式接入，計劃在 11 月底能夠接入 20 億事件量 / 日。

智聯(lián)招聘也在持續(xù)為 Apache Pulsar 社區(qū)貢獻新的特性比如 Dead Letter Topic， Client Interceptors 等，智聯(lián)招聘有很多業(yè)務(wù)場景也非常依賴延時消息特性，后面我們也會在 Pulsar 上貢獻此特性。

感謝 Apache Pulsar 社區(qū)小伙伴們在項目落地過程中的技術(shù)支持。

標簽： Mysql ssd 安全 數(shù)據(jù)庫

版權(quán)申明：本站文章部分自網(wǎng)絡(luò)，如有侵權(quán)，請聯(lián)系：west999com@outlook.com
特別注意：本站所有轉(zhuǎn)載文章言論不代表本站觀點！
本站所提供的圖片等素材，版權(quán)歸原作者所有，如需使用，請與原作者聯(lián)系。