基于Queue + Stream的統一消息消費模型是怎么樣的

基于Queue + Stream的統一消息消費模型是怎么樣的，很多新手對此不是很清楚，為了幫助大家解決這個難題，下面小編將為大家詳細講解，有這方面需求的人可以來學習下，希望你能有所收獲。

之前的文章，我們描述了Apache Pulsar能夠成為企業級流和消息系統的原因。Pulsar的企業特性包括消息的持久化存儲，多租戶，多機房互聯互備，加密和安全性等。我們經常被問到的一個問題是Apache Pulsar和Apache Kafka有什么不同。

在用戶選擇一個消息系統時，消息模型是用戶首先考慮的事情。消息模型應涵蓋以下3個方面：

1. 消息消費 - 如何發送和消費消息；

2. 消息確認（ack） - 如何確認消息；

3. 消息保存 - 消息保留多長時間，觸發消息刪除的原因以及怎樣刪除；

消息消費模型

在實時流式架構中，消息傳遞可以分為兩類：隊列（Queue）和流（Stream）。

隊列（Queue）模型

隊列模型主要是采用無序或者共享的方式來消費消息。通過隊列模型，用戶可以創建多個消費者從單個管道中接收消息；當一條消息從隊列發送出來后，多個消費者中的只有一個（任何一個都有可能）接收和消費這條消息。消息系統的具體實現決定了最終哪個消費者實際接收到消息。

隊列模型通常與無狀態應用程序一起結合使用。無狀態應用程序不關心排序，但它們確實需要能夠確認（ack）或刪除單條消息，以及盡可能地擴展消費并行性的能力。典型的基于隊列模型的消息系統包括RabbitMQ和RocketMQ。

流式（Stream）模型

相比之下，流模型要求消息的消費嚴格排序或獨占消息消費。對于一個管道，使用流式模型，始終只會有一個消費者使用和消費消息。消費者按照消息寫入管道的確切順序接收從管道發送的消息。

流模型通常與有狀態應用程序相關聯。有狀態的應用程序更加關注消息的順序及其狀態。消息的消費順序決定了有狀態應用程序的狀態。消息的順序將影響應用程序處理邏輯的正確性。

在面向微服務或事件驅動的體系結構中，隊列模型和流模型都是必需的。

Pulsar的消息消費模型

Apache Pulsar通過“訂閱”，抽象出了統一的: producer-topic-subscription-consumer 消費模型。Pulsar的消息模型既支持隊列模型，也支持流模型。

在Pulsar的消息消費模型中，Topic是用于發送消息的通道。每一個Topic對應著Apache BookKeeper中的一個分布式日志。發布者發布的每條消息只在Topic中存儲一次；存儲的過程中，BookKeeper會將消息復制存儲在多個存儲節點上；Topic中的每條消息，可以根據消費者的訂閱需求，多次被使用，每個訂閱對應一個消費者組（Consumer Group）。

主題（Topic）是消費消息的真實來源。盡管消息僅在主題（Topic）上存儲一次，但是用戶可以有不同的訂閱方式來消費這些消息：

• 消費者被組合在一起以消費消息，每個消費組是一個訂閱。

• 每個Topic可以有不同的消費組。

• 每組消費者都是對主題的一個訂閱。

• 每組消費者可以擁有自己不同的消費方式： 獨占（Exclusive），故障切換（Failover）或共享（Share）。
  
  

Pulsar通過這種模型，將隊列模型和流模型這兩種模型結合在了一起，提供了統一的API接口。 這種模型，既不會影響消息系統的性能，也不會帶來額外的開銷，同時還為用戶提供了更多靈活性，方便用戶程序以最匹配模式來使用消息系統。

獨占訂閱（Stream流模型）

顧名思義，獨占訂閱中，在任何時間，一個消費者組（訂閱）中有且只有一個消費者來消費Topic中的消息。下圖是獨占訂閱的示例。在這個示例中有一個有訂閱A的活躍消費者A-0，消息m0到m4按順序傳送并由A-0消費。如果另一個消費者A-1想要附加到訂閱A，則是不被允許的。

故障切換（Stream流模型）

使用故障切換訂閱，多個消費者（Consumer）可以附加到同一訂閱。 但是，一個訂閱中的所有消費者，只會有一個消費者被選為該訂閱的主消費者。 其他消費者將被指定為故障轉移消費者。
當主消費者斷開連接時，分區將被重新分配給其中一個故障轉移消費者，而新分配的消費者將成為新的主消費者。 發生這種情況時，所有未確認（ack）的消息都將傳遞給新的主消費者。 這類似于Apache Kafka中的Consumer partition rebalance。
下圖是故障切換訂閱的示例。 消費者B-0和B-1通過訂閱B訂閱消費消息。B-0是主消費者并接收所有消息。 B-1是故障轉移消費者，如果消費者B-0出現故障，它將接管消費。

共享訂閱（Queue隊列模型）

使用共享訂閱，在同一個訂閱背后，用戶按照應用的需求掛載任意多的消費者。 訂閱中的所有消息以循環分發形式發送給訂閱背后的多個消費者，并且一個消息僅傳遞給一個消費者。
當消費者斷開連接時，所有傳遞給它但是未被確認（ack）的消息將被重新分配和組織，以便發送給該訂閱上剩余的剩余消費者。
下圖是共享訂閱的示例。 消費者C-1，C-2和C-3都在同一主題上消費消息。 每個消費者接收大約所有消息的1/3。
如果想提高消費的速度，用戶不需要不增加分區數量，只需要在同一個訂閱中添加更多的消費者。

三種訂閱模式的選擇

獨占和故障切換訂閱，僅允許一個消費者來使用和消費，每個對主題的訂閱。這兩種模式都按主題分區順序使用消息。它們最適用于需要嚴格消息順序的流（Stream）用例。

共享訂閱允許每個主題分區有多個消費者。同一訂閱中的每個消費者僅接收主題分區的一部分消息。共享訂閱最適用于不需要保證消息順序的隊列（Queue）的使用模式，并且可以按照需要任意擴展消費者的數量。

Pulsar中的訂閱實際上與Apache Kafka中的Consumer Group的概念類似。創建訂閱的操作很輕量化，而且具有高度可擴展性，用戶可以根據應用的需要創建任意數量的訂閱。
對同一主題的不同訂閱，也可以采用不同的訂閱類型。比如用戶可以在同一主題上可以提供一個包含3個消費者的故障切換訂閱，同時也提供一個包含20個消費者的共享訂閱，并且可以在不改變分區數量的情況下，向共享訂閱添加更多的消費者。
下圖描繪了一個包含3個訂閱A，B和C的主題，并說明了消息如何從生產者流向消費者。

除了統一消息API之外，由于Pulsar主題分區實際上是存儲在Apache BookKeeper中，它還提供了一個讀取API（Reader），類似于消費者API（但Reader沒有游標管理），以便用戶完全控制如何使用Topic中的消息。

Pulsar的消息確認（ACK）

由于分布式系統的特性，當使用分布式消息系統時，可能會發生故障。比如在消費者從消息系統中的主題消費消息的過程中，消費消息的消費者和服務于主題分區的消息代理（Broker）都可能發生錯誤。消息確認（ACK）的目的就是保證當發生這樣的故障后，消費者能夠從上一次停止的地方恢復消費，保證既不會丟失消息，也不會重復處理已經確認（ACK）的消息。
在Apache Kafka中，恢復點通常稱為Offset，更新恢復點的過程稱為消息確認或提交Offset。

在Apache Pulsar中，每個訂閱中都使用一個專門的數據結構--游標（Cursor）來跟蹤訂閱中的每條消息的確認（ACK）狀態。每當消費者在主題分區上確認消息時，游標都會更新。更新游標可確保消費者不會再次收到消息。

Apache Pulsar提供兩種消息確認方法，單條確認（Individual Ack）和累積確認（Cumulative Ack）。通過累積確認，消費者只需要確認它收到的最后一條消息。主題分區中的所有消息（包括）提供消息ID將被標記為已確認，并且不會再次傳遞給消費者。累積確認與Apache Kafka中的Offset更新類似。

Apache Pulsar可以支持消息的單條確認，也就是選擇性確認。消費者可以單獨確認一條消息。 被確認后的消息將不會被重新傳遞。下圖說明了單條確認和累積確認的差異（灰色框中的消息被確認并且不會被重新傳遞）。在圖的上半部分，它顯示了累計確認的一個例子，M12之前的消息被標記為acked。在圖的下半部分，它顯示了單獨進行acking的示例。僅確認消息M7和M12 - 在消費者失敗的情況下，除了M7和M12之外，其他所有消息將被重新傳送。

獨占訂閱或故障切換訂閱的消費者能夠對消息進行單條確認和累積確認；共享訂閱的消費者只允許對消息進行單條確認。單條確認消息的能力為處理消費者故障提供了更好的體驗。對于某些應用來說，處理一條消息可能需要很長時間或者非常昂貴，防止重新傳送已經確認的消息非常重要。

這個管理Ack的專門的數據結構--游標（Cursor），由Broker來管理，利用BookKeeper的Ledger提供存儲，在后面的文章中我們會介紹更多的關于游標（Cursor）的細節。

Apache Pulsar提供了靈活的消息消費訂閱類型和消息確認方法，通過簡單的統一的API，就可以支持各種消息和流的使用場景。

Pulsar的消息保留（Retention）

在消息被確認后，Pulsar的Broker會更新對應的游標。當Topic里面中的一條消息，被所有的訂閱都確認ack后，才能刪除這條消息。Pulsar還允許通過設置保留時間，將消息保留更長時間，即使所有訂閱已經確認消費了它們。
下圖說明了如何在有2個訂閱的主題中保留消息。訂閱A在M6和訂閱B已經消耗了M10之前的所有消息之前已經消耗了所有消息。這意味著M6之前的所有消息（灰色框中）都可以安全刪除。訂閱A仍未使用M6和M9之間的消息，無法刪除它們。如果主題配置了消息保留期，則消息M0到M5將在配置的時間段內保持不變，即使A和B已經確認消費了它們。

在消息保留策略中，Pulsar還支持消息生存時間（TTL）。如果消息未在配置的TTL時間段內被任何消費者使用，則消息將自動標記為已確認。 消息保留期消息TTL之間的區別在于：消息保留期作用于標記為已確認并設置為已刪除的消息，而TTL作用于未ack的消息。 上面的圖例中說明了Pulsar中的TTL。 例如，如果訂閱B沒有活動消費者，則在配置的TTL時間段過后，消息M10將自動標記為已確認，即使沒有消費者實際讀取該消息。

Pulsar VS. Kafka

通過以上幾個方面，我們對Pulsar和Kafka在消息模型方面的不同點進行一個總結。

模型概念

Kafka： Producer - topic - consumer group - consumer；

Pulsar：Producer - topic - subscription - consumer。

消費模式

Kafka： 主要集中在流（Stream）模式，對單個partition是獨占消費，沒有共享（Queue）的消費模式；

Pulsar：提供了統一的消息模型和API。流（Stream）模式 -- 獨占和故障切換訂閱方式；隊列（Queue）模式 -- 共享訂閱的方式。

消息確認（Ack）

Kafka： 使用偏移Offset；

Pulsar：使用專門的Cursor管理。累積確認和Kafka效果一樣；提供單條或選擇性確認。

消息保留
Kafka：根據設置的保留期來刪除消息。有可能消息沒被消費，過期后被刪除。 不支持TTL。

Pulsar：消息只有被所有訂閱消費后才會刪除，不會丟失數據。也允許設置保留期，保留被消費的數據。支持TTL。

對比總結：

Apache Pulsar將高性能的流（Apache Kafka所追求的）和靈活的傳統隊列（RabbitMQ所追求的）結合到一個統一的消息模型和API中。 Pulsar使用統一的API為用戶提供一個支持流和隊列的系統，且具有同樣的高性能。

看完上述內容是否對您有幫助呢？如果還想對相關知識有進一步的了解或閱讀更多相關文章，請關注億速云行業資訊頻道，感謝您對億速云的支持。