關(guān)鍵要點(diǎn)

• 分布式消息傳遞系統(tǒng)支持兩種類型的語義：流式傳輸和排隊(duì)。每個都最適合某些類型的用例。
• Apache Pulsar 的獨(dú)特之處在于它同時支持流式處理和排隊(duì)用例。
• Pulsar 的多層架構(gòu)允許用戶比其他消息傳遞系統(tǒng)更方便地擴(kuò)展主題的數(shù)量和大小。
• Pulsar 提供了可擴(kuò)展性、可靠性和功能的適當(dāng)平衡，以取代 Iterable 的 RabbitMQ，并最終取代其他消息傳遞系統(tǒng)，如 Kafka 和 Amazon SQS。

在 Iterable，我們每天代表客戶發(fā)送大量營銷信息。這些包括電子郵件、推送、SMS 和應(yīng)用內(nèi)消息。Iterable 還每天處理更多的用戶更新、事件和自定義工作流狀態(tài)，其中許多可以觸發(fā)系統(tǒng)中的其他操作。這導(dǎo)致系統(tǒng)不僅對我們的客戶非常有用，而且非常復(fù)雜。隨著我們客戶群的增長，管理這種復(fù)雜性變得更加重要。

Iterable 管理復(fù)雜性的一種方法是在其架構(gòu)的幾個部分中使用分布式消息傳遞系統(tǒng)。分布式消息系統(tǒng)的主要目的是存儲需要消費(fèi)者處理的消息，并在處理這些消息時跟蹤這些消費(fèi)者的狀態(tài)。這樣，消費(fèi)者可以專注于處理每條消息的任務(wù)。

Iterable 使用工作隊(duì)列方法來執(zhí)行客戶指定的營銷工作流、webhook 和其他類型的作業(yè)調(diào)度和處理。其他組件，例如用戶和事件攝取，使用流模型來處理有序的消息流。

通常，分布式消息傳遞系統(tǒng)支持兩種類型的語義：流式傳輸和排隊(duì)。每個都最適合某些類型的用例。

流式傳輸和排隊(duì)

在流式消息系統(tǒng)中，生產(chǎn)者將數(shù)據(jù)附加到一組僅附加的消息流中。在每個流中，消息必須按特定順序處理，并且消費(fèi)者在流中標(biāo)記它們的位置?？梢允褂媚撤N策略（例如散列用戶 ID）對消息進(jìn)行分區(qū)以允許更大的并行性，并且每個分區(qū)都充當(dāng)單獨(dú)的數(shù)據(jù)流。因?yàn)槊總€流中的數(shù)據(jù)都是不可變的，并且只存儲了偏移量條目，所以消息可能不會被跳過。流式處理在消息順序很重要的情況下效果很好，例如數(shù)據(jù)攝取。Kafka和Amazon Kinesis是使用流語義來消費(fèi)消息的消息系統(tǒng)示例。

在隊(duì)列消息系統(tǒng)中，生產(chǎn)者將消息發(fā)送到可能由多個消費(fèi)者共享的隊(duì)列。消費(fèi)者在收到消息時對其進(jìn)行處理，并在處理每條消息時向排隊(duì)系統(tǒng)發(fā)送確認(rèn)。因?yàn)槎鄠€消費(fèi)者可能共享一個隊(duì)列并且消息序列并不重要，所以通常更容易擴(kuò)展基于隊(duì)列的系統(tǒng)的消費(fèi)者端。排隊(duì)系統(tǒng)非常適合不需要按特定順序執(zhí)行任務(wù)的工作隊(duì)列——例如，向多個收件人發(fā)送一封電子郵件。RabbitMQ和Amazon SQS是流行的基于隊(duì)列的消息系統(tǒng)的示例。

排隊(duì)系統(tǒng)通常包括簡化處理消息級錯誤任務(wù)的功能。例如，發(fā)生錯誤后，RabbitMQ 可以輕松地將消息傳輸?shù)教厥怅?duì)列，在該隊(duì)列中保留指定的時間，然后返回到原始隊(duì)列進(jìn)行重試。它還可以否定地確認(rèn)一條消息，以便在失敗后重新傳遞它。由于大多數(shù)消息隊(duì)列在確認(rèn)消息后通常不會將消息存儲在積壓中，因此調(diào)試和災(zāi)難恢復(fù)更加困難，因?yàn)闆]有要檢查的消息。

像 Kafka 這樣的基于流的系統(tǒng)可用于排隊(duì)用例，但有一些注意事項(xiàng)。事實(shí)上，許多用戶選擇此選項(xiàng)是因?yàn)檫@些系統(tǒng)通常提供卓越的性能。然而，這種解決方案可能是一個挑戰(zhàn)，因?yàn)樗o開發(fā)人員帶來了過度的負(fù)擔(dān)，無法處理嚴(yán)格的流排序所帶來的限制。如果消費(fèi)者消費(fèi)消息的速度很慢或需要在暫時失敗后重試處理，則同一流上其他消息的處理可能會延遲。一個常見的解決方案是通過將消息重新發(fā)布到另一個主題來重試處理，但這會引入復(fù)雜性，因?yàn)閼?yīng)用程序邏輯必須管理額外的狀態(tài)。

為什么 Iterable 需要一個新的消息傳遞平臺

我們一直在大量使用 RabbitMQ，并依靠它的特性來處理內(nèi)部消息傳遞。我們大量使用生存時間 (TTL) 值，不僅用于固定長度的重試，還用于在消息處理中實(shí)現(xiàn)顯式延遲。例如，我們可能會延遲發(fā)送營銷電子郵件，以便營銷信息可以在每個收件人最有可能打開的時間發(fā)送給他們。我們還依靠否定確認(rèn)來重試排隊(duì)的消息。

這是我們架構(gòu)的簡化版本：

當(dāng)我們開始評估 Pulsar 時，上面提到的所有隊(duì)列都在 RabbitMQ 上，除了使用 Kafka 的攝取。Kafka 非常適合攝取，因?yàn)樗峁┝吮匾男阅芎团判虮ＷC。Kafka 不適合其他用例，因?yàn)樗狈Ρ匾墓ぷ麝?duì)列語義。我們使用了許多 RabbitMQ 特有的特性，比如延遲，這一事實(shí)也使得尋找替代方案變得更具挑戰(zhàn)性。

當(dāng)我們擴(kuò)展我們的系統(tǒng)時，RabbitMQ 開始出現(xiàn)以下限制：

• 在高負(fù)載下，RabbitMQ 經(jīng)常遇到流量控制問題。流控制是一種在消息代理無法跟上時減慢發(fā)布者的機(jī)制，通常是因?yàn)閮?nèi)存和其他資源限制。這阻礙了生產(chǎn)者發(fā)布的能力，從而導(dǎo)致其他領(lǐng)域的服務(wù)延遲和請求失敗。具體來說，我們注意到當(dāng)大量消息的 TTL 同時過期時，流控制更頻繁地發(fā)生。在這些情況下，RabbitMQ 試圖一次性將過期消息傳遞到它們的目標(biāo)隊(duì)列。這使 RabbitMQ 實(shí)例的內(nèi)存容量不堪重負(fù)，從而觸發(fā)了正常生產(chǎn)者的流控機(jī)制，阻止了他們的發(fā)布嘗試。
• 調(diào)試變得更加困難，因?yàn)?RabbitMQ 的代理在消息被確認(rèn)后不存儲消息。換言之，無法為消息設(shè)置保留時間。
• 復(fù)制很難實(shí)現(xiàn)，因?yàn)?RabbitMQ 中的復(fù)制組件對于我們的用例來說不夠健壯，導(dǎo)致 RabbitMQ 成為我們消息狀態(tài)的單點(diǎn)故障。
• RabbitMQ 難以處理大量隊(duì)列。由于我們有許多需要專用隊(duì)列的用例，我們通常一次需要超過 10,000 個隊(duì)列。在這個級別，RabbitMQ 遇到了性能問題，通常首先出現(xiàn)在管理界面和 API 中。

評估 Apache Pulsar

總的來說，Apache Pulsar似乎提供了我們需要的所有功能。雖然我們在 Pulsar 周圍看到的很多宣傳都將其與 Kafka 進(jìn)行流式工作負(fù)載進(jìn)行比較，但我們也發(fā)現(xiàn) Pulsar 非常適合我們的排隊(duì)需求。Pulsar 的共享訂閱功能允許將主題用作隊(duì)列，可能為同一主題中的不同訂閱者提供多個虛擬隊(duì)列。Pulsar 還原生支持延遲和預(yù)定消息，盡管這些功能在我們開始考慮 Pulsar 時是非常新的。

除了提供豐富的功能集之外，Pulsar 的多層架構(gòu)讓我們能夠比其他消息傳遞系統(tǒng)更方便地擴(kuò)展主題的數(shù)量和大小。

Pulsar 的頂層由代理組成，它們接受來自生產(chǎn)者的消息并將它們發(fā)送給消費(fèi)者，但不存儲數(shù)據(jù)。單個代理處理每個主題分區(qū)，但代理可以輕松交換主題所有權(quán)，因?yàn)樗鼈儾淮鎯χ黝}狀態(tài)。這使得添加代理以增加吞吐量并立即利用新代理變得容易。這也使 Pulsar 能夠處理代理故障。

Pulsar 的底層BookKeeper將主題數(shù)據(jù)存儲在分段中，這些分段分布在整個集群中。如果需要額外的存儲空間，我們可以輕松地將 BookKeeper 節(jié)點(diǎn)（bookies）添加到集群中，并使用它們來存儲新的主題段。經(jīng)紀(jì)人與博彩公司協(xié)調(diào)，以在每個主題發(fā)生變化時更新它的狀態(tài)。Pulsar 將 BookKeeper 用于主題數(shù)據(jù)也有助于它支持大量主題，這對于 Iterable 當(dāng)前的許多用例至關(guān)重要。

在評估了幾個消息傳遞系統(tǒng)之后，我們認(rèn)為 Pulsar 提供了可擴(kuò)展性、可靠性和功能之間的適當(dāng)平衡，以取代 Iterable 的 RabbitMQ，并最終取代其他消息傳遞系統(tǒng)，如 Kafka 和 Amazon SQS。

第一個 Pulsar 用例：消息發(fā)送

Iterable 平臺最重要的功能之一是代表 Iterable 的客戶安排和發(fā)送營銷電子郵件。為此，我們將消息發(fā)布到客戶特定的隊(duì)列，然后讓另一個服務(wù)處理消息的最終呈現(xiàn)和發(fā)送。這些隊(duì)列是我們決定從 RabbitMQ 遷移到 Pulsar 的第一件事。

我們選擇營銷信息發(fā)送作為我們的第一個 Pulsar 用例有兩個原因。首先，因?yàn)榘l(fā)送包含了我們一些更復(fù)雜的 RabbitMQ 用例。其次，因?yàn)樗砹宋覀?RabbitMQ 使用的很大一部分。這不是風(fēng)險最低的用例；然而，經(jīng)過廣泛的性能和可擴(kuò)展性測試，我們認(rèn)為這是 Pulsar 可以增加最大價值的地方。

以下是在 Iterable 平臺上創(chuàng)建的三種常見活動類型：

1. 同時向所有收件人發(fā)送營銷信息的爆炸式營銷活動。假設(shè)客戶想要向過去一個月活躍的用戶發(fā)送電子郵件通訊。在這種情況下，我們可以在計(jì)劃活動時查詢ElasticSearch的用戶列表，并將它們發(fā)布到該客戶的 Pulsar 主題。
2. 為每個收件人指定自定義發(fā)送時間的 Blast 營銷活動。發(fā)送時間可以是固定的（例如，“收件人當(dāng)?shù)貢r區(qū)的上午 9 點(diǎn)”），也可以由我們的發(fā)送時間優(yōu)化功能計(jì)算得出。在每種情況下，我們都希望將排隊(duì)消息的處理延遲到指定時間。
3. 用戶觸發(fā)的廣告系列。這些可以由自定義工作流程或用戶發(fā)起的交易（例如在線購買）觸發(fā)。用戶觸發(fā)的營銷發(fā)送是根據(jù)需要單獨(dú)完成的。

在上述每種情況下，在任何給定時間執(zhí)行的發(fā)送數(shù)量可能會有很大差異，因此我們還需要能夠向上和向下擴(kuò)展消費(fèi)者以適應(yīng)不斷變化的負(fù)載。

遷移到 Apache Pulsar

盡管 Pulsar 在負(fù)載測試中表現(xiàn)良好，但我們不確定它是否能夠在生產(chǎn)中維持高負(fù)載水平。這是一個特別關(guān)注的問題，因?yàn)槲覀冇?jì)劃利用 Pulsar 的幾個新功能，包括否定確認(rèn)和預(yù)定消息傳遞。

為了建立我們的信心，我們實(shí)現(xiàn)了一個并行管道，在該管道中我們向 RabbitMQ 和 Pulsar 發(fā)布消息；在這種情況下，我們在這些主題上設(shè)置消費(fèi)者來確認(rèn)排隊(duì)的消息而不實(shí)際處理它們。我們還模擬了消費(fèi)延遲。這有助于我們了解 Pulsar 在特定生產(chǎn)環(huán)境中的行為。我們對測試主題和實(shí)際生產(chǎn)主題都使用了客戶級別的功能標(biāo)志，因此我們可以逐一遷移客戶以進(jìn)行測試，并最終用于生產(chǎn)用途。

在測試過程中，我們發(fā)現(xiàn)了 Pulsar 中的一些錯誤。例如，我們發(fā)現(xiàn)了與延遲消息相關(guān)的競爭條件，Pulsar 開發(fā)人員幫助識別和修復(fù)了這種情況。這是我們發(fā)現(xiàn)的最嚴(yán)重的問題，因?yàn)樗鼘?dǎo)致消費(fèi)者卡住，造成未消費(fèi)消息的積壓。

我們還注意到一些與 Pulsar 消息批處理相關(guān)的有趣問題，這些問題在 Pulsar 生產(chǎn)者中默認(rèn)啟用。例如，我們注意到 Pulsar 的積壓指標(biāo)報告的是批次數(shù)，而不是實(shí)際的消息數(shù)量，這使得為消息積壓設(shè)置警報閾值更具挑戰(zhàn)性。后來我們在否定確認(rèn)和批處理之間的交互中發(fā)現(xiàn)了一個更嚴(yán)重的錯誤，最近已經(jīng)修復(fù)了。最終，我們認(rèn)為批處理不值得麻煩。幸運(yùn)的是，在 Pulsar 生產(chǎn)者中禁用批處理很容易，并且沒有批處理的性能足以滿足我們的需求。這些問題也可能在即將發(fā)布的版本中得到修復(fù)。

延遲和否定確認(rèn)在當(dāng)時是相對較新的功能，因此我們預(yù)計(jì)我們可能會發(fā)現(xiàn)一些問題。這就是為什么我們選擇在幾個月內(nèi)緩慢遷移到 Pulsar，最初發(fā)布僅用于測試主題，然后逐漸遷移真實(shí)發(fā)送。這種方法使我們能夠在問題成為我們客戶的問題之前發(fā)現(xiàn)問題。盡管花了大約六個月的時間才完全相信 Pulsar 正在按預(yù)期工作，但結(jié)果值得花時間。

在大約六個月的時間里，我們將整個營銷發(fā)送操作遷移到了 Pulsar。遷移完成后，我們發(fā)現(xiàn) Pulsar 將我們的運(yùn)營成本降低了近一半，并且在我們增加新客戶時還有增長空間。成本降低顯著，部分原因是我們的 RabbitMQ 實(shí)例被過度配置以彌補(bǔ)性能問題。迄今為止，我們的 Pulsar 集群已經(jīng)順利運(yùn)行了六個多月，沒有出現(xiàn)任何問題。

實(shí)施和工具

Iterable 主要在后端使用Scala，因此為 Pulsar 提供良好的 Scala 工具對我們來說很重要。我們使用了優(yōu)秀的pulsar4s庫，并做出了許多支持新功能的貢獻(xiàn)，例如延遲消息。我們還提供了一個基于 Akka Streams 的連接器，用于將消息用作源，并提供單獨(dú)的確認(rèn)支持。

例如，我們可以像這樣使用命名空間中的所有主題：

// Create a consumer on all topics in this namespace
val createConsumer = () => client.consumer(ConsumerConfig(
  topicPattern = "persistent://email/project-123/.*".r,
  subscription = Subscription("email-service")
))
// Create an Akka streams `Source` stage for this consumer
val pulsarSource = committableSource(createConsumer, Some(MessageId.earliest))
// Materialize the source and get back a `control` to shut it down later.
val control = pulsarSource.mapAsync(parallelism)(handleMessage).to(Sink.ignore).run()

我們喜歡為消費(fèi)者使用正則表達(dá)式訂閱。它們使在創(chuàng)建新主題時自動訂閱它們變得容易，并使消費(fèi)者不必了解特定的主題分區(qū)策略。同時，我們也在利用 Pulsar 支持大量主題的能力。由于 Pulsar 會在發(fā)布時自動創(chuàng)建新主題，因此為新消息類型甚至單個活動創(chuàng)建新主題很簡單。這也使得為不同的客戶和消息類型實(shí)施速率限制變得更加容易。

我們學(xué)到了什么

由于 Pulsar 是一個快速發(fā)展的開源項(xiàng)目，我們遇到了一些挑戰(zhàn)——主要是在加快速度和學(xué)習(xí)它的怪癖方面——我們可能不會在其他更成熟的技術(shù)中看到這些挑戰(zhàn)。文檔并不總是完整的，我們經(jīng)常需要向社區(qū)尋求幫助。也就是說，社區(qū)非常熱情和樂于助人，我們很高興更多地參與 Pulsar 的開發(fā)并參與有關(guān)新功能的討論。

Pulsar 的獨(dú)特之處在于它同時支持流式傳輸和排隊(duì)用例，同時還支持廣泛的功能集，使其成為我們架構(gòu)中當(dāng)前使用的許多其他分布式消息傳遞技術(shù)的可行替代方案。Pulsar 涵蓋了我們所有的 Kafka、RabbitMQ 和 SQS 用例。這使我們能夠?qū)Ｗ⒂趪@單個統(tǒng)一系統(tǒng)構(gòu)建專業(yè)知識和工具。

自 2019 年初開始與 Pulsar 合作以來，我們一直對 Pulsar 的開發(fā)進(jìn)展感到鼓舞，尤其是在初學(xué)者的準(zhǔn)入門檻方面。工具有了很大的改進(jìn)：例如，Pulsar Manager現(xiàn)在提供了一個非常方便的 GUI 來管理集群。我們還看到許多公司提供托管和托管的 Pulsar 服務(wù)，這使初創(chuàng)公司和小型團(tuán)隊(duì)更容易開始使用 Pulsar。

總體而言，Iterable 向 Pulsar 的過渡很有趣，有時也很有挑戰(zhàn)性，但到目前為止相當(dāng)成功。在許多方面，我們的用例代表了一條尚未被廣泛采用的新路徑。我們預(yù)計(jì)會遇到一些問題，但我們的測試過程有助于最大限度地減少它們對客戶的影響。我們現(xiàn)在對使用 Pulsar 充滿信心，并將繼續(xù)擴(kuò)展我們對 Pulsar 的使用，用于 Iterable 平臺中的其他現(xiàn)有和新組件。