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

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

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

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

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

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

流式傳輸和排隊(duì)

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

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

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

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

為什么 Iterable 需要一個(gè)新的消息傳遞平臺(tái)

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

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

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

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

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

評(píng)估 Apache Pulsar

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

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

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

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

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

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

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

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

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

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

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

遷移到 Apache Pulsar

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

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

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

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

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

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

實(shí)施和工具

Iterable 主要在后端使用Scala，因此為 Pulsar 提供良好的 Scala 工具對(duì)我們來(lái)說(shuō)很重要。我們使用了優(yōu)秀的pulsar4s庫(kù)，并做出了許多支持新功能的貢獻(xiàn)，例如延遲消息。我們還提供了一個(gè)基于 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)建新主題時(shí)自動(dòng)訂閱它們變得容易，并使消費(fèi)者不必了解特定的主題分區(qū)策略。同時(shí)，我們也在利用 Pulsar 支持大量主題的能力。由于 Pulsar 會(huì)在發(fā)布時(shí)自動(dòng)創(chuàng)建新主題，因此為新消息類型甚至單個(gè)活動(dòng)創(chuàng)建新主題很簡(jiǎn)單。這也使得為不同的客戶和消息類型實(shí)施速率限制變得更加容易。

我們學(xué)到了什么

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

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

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

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