（四）——消費者的競爭模式

云計算設(shè)計模式（四）——消費者的競爭模式

允許多個并發(fā)用戶處理在同一個通訊通道接收的消息。這種模式使系統(tǒng)能夠同時處理多個郵件，以優(yōu)化吞吐量，提高可擴展性和可用性，以及平衡工作負載。

背景和問題

在云中運行的應(yīng)用程序，可以預(yù)計，以處理大量的請求。而不是過程的每個請求同步地，一個常用的方法是通過一個消息傳送系統(tǒng)到該異步地處理它們的另一服務(wù)（消費者服務(wù)），以通過他們的應(yīng)用程序。這種策略有助于確保在應(yīng)用程序的業(yè)務(wù)邏輯沒有被阻塞，而正在處理的請求。

請求的數(shù)量可以隨著時間的原因有很多顯著變化。突然一陣在用戶活動或聚集的請求，來自多個租戶未來可能會導(dǎo)致不可預(yù)測的工作負載。在高峰時間的系統(tǒng)可能需要處理許多每秒數(shù)百個請求，而在其他時間的數(shù)量可能是非常小的。此外，該工作的性質(zhì)進行的處理這些請求可能是高度可變的。使用消費者服務(wù)的單個實例，可能會導(dǎo)致該實例成為充斥請求或消息傳送系統(tǒng)可通過消息從應(yīng)用程序來的流入被重載。為了處理這種波動的負載，該系統(tǒng)可以運行消費者服務(wù)的多個實例。然而這些消費者必須協(xié)調(diào)，以確保每個消息只傳送給一個單個消費者。工作量也需要跨消費者被負載平衡，以防止一個實例成為瓶頸。

解決方案

使用消息隊列來實現(xiàn)應(yīng)用和消費者服務(wù)的實例之間的通信信道。在消息隊列中的形式應(yīng)用帖請求，以及消費者的服務(wù)實例從隊列中接收消息并對其進行處理。這種方法使消費者的服務(wù)實例的同一池中從應(yīng)用程序的任何實例處理消息。圖 1 示出了該架構(gòu)。

圖1 - 使用消息隊列分發(fā)工作提高到一個服務(wù)的實例

該解決方案具有以下優(yōu)點：

• 它使固有的負載調(diào)平系統(tǒng)，可以處理由應(yīng)用程序?qū)嵗l(fā)送請求量很大的變化。隊列充當(dāng)應(yīng)用程序?qū)嵗拖M者服務(wù)實例，這有助于最大限度地減少對應(yīng)用程序和服務(wù)實例（所描述的基于隊列的負載調(diào)平模式）的可用性和響應(yīng)性的影響之間的緩沖區(qū)。處理的消息，需要一些將被執(zhí)行時，不會妨礙同時由消費者服務(wù)的其他實例所處理的其它消息長期運行的處理。
• 它提高了可靠性。如果一個生產(chǎn)者直接與消費者，而不是使用這種模式進行通信，但不監(jiān)視消費者，有一個高概率的消息可能丟失或失敗，如果消費者無法進行處理。在這種模式的消息不被發(fā)送給一個特定的服務(wù)實例，一個失敗服務(wù)實例不會阻塞一個生產(chǎn)者和消息可以通過任何加工服務(wù)實例進行處理。
• 它不需要復(fù)雜的協(xié)調(diào)的消費者之間，或在生產(chǎn)者和消費者的實例。消息隊列確保每個消息傳遞至少一次。
• 它是可擴展的。該系統(tǒng)能夠動態(tài)地增加或減少消費者服務(wù)的實例的數(shù)目的消息量是波動的。
• 它可以提高彈性，如果消息隊列提供事務(wù)讀取操作。如果消費者服務(wù)實例能夠讀取和處理該消息作為一個事務(wù)操作的一部分，并且如果這種消費服務(wù)實例隨后發(fā)生故障時，這種模式可以確保該消息將被返回到隊列中被拾起并處理通過的另一個實例消費者服務(wù)。

問題和注意事項

在決定如何實現(xiàn)這個模式時，請考慮以下幾點：

• 留言訂購。其中消費者服務(wù)實例接收消息的順序是無法保證的，并且不一定反映了所創(chuàng)建的消息中的順序。設(shè)計系統(tǒng)，以確保信息的處理是冪等的，因為這將有助于消除該消息的處理順序上的任何依賴。有關(guān)冪等的詳細信息，請參閱喬納森·奧利弗的博客冪等模式??。

注意

微軟 Azure 服務(wù)總線隊列可以通過使用消息會先入先出消息的順序工具保證。欲了解更多信息，請參閱消息傳遞模式 MSDN 上使用會話。

• 設(shè)計服務(wù)的永續(xù)性。如果系統(tǒng)被設(shè)計為檢測和重新啟動失敗的服務(wù)實例中，可能有必要執(zhí)行由服務(wù)??實例執(zhí)行作為冪等操作，以最小化被檢索和處理一次以上的單個消息的影響的處理。
• 檢測有害消息。格式不正確的消息，或者需要訪問不可用的資源的任務(wù)，可能會造成服務(wù)實例失敗。該系統(tǒng)應(yīng)避免這樣的消息被返回到隊列，而是捕獲和別處存儲這些信息的詳細信息，以便可以在需要進行分析。
• 處理結(jié)果。服務(wù)實例處理一個消息從生成該消息的應(yīng)用程序邏輯完全分離，并且它們未必能夠直接進行通信。如果服務(wù)實例生成必須傳回給應(yīng)用程序邏輯結(jié)果，該信息必須被存儲在一個位置，都可以訪問兩個和系統(tǒng)必須提供某種指示時的處理已經(jīng)完成，以防止應(yīng)用邏輯從檢索數(shù)據(jù)不全。

注意

如果您正在使用 Azure 的工作進程可能能夠通過使用專用的郵件回復(fù)隊列回傳結(jié)果的應(yīng)用程序邏輯。應(yīng)用邏輯必須能夠?qū)⑦@些結(jié)果與原來的消息關(guān)聯(lián)起來。這種情況下進行了更詳細的異步消息的引物進行說明。

• 擴展的信息系統(tǒng)。在一個大型的解決方案，一個消息隊列可以是不堪重負的消息的數(shù)量，并成為系統(tǒng)中的瓶頸。在這種情況下，考慮分割該消息系統(tǒng)直接從特定制造商的信息到一個特定的隊列，或使用負載平衡，以跨多個消息隊列分發(fā)消息。
• 郵件系統(tǒng)的可靠性保障。一個可靠的消息傳送系統(tǒng)，需要保證的是，一旦應(yīng)用程序放入隊列的消息，它也不會丟失。這是確保所有郵件傳遞至少一次重要的。

當(dāng)使用這個模式

使用這種模式時：

• 工作量為一個應(yīng)用程序被分成可異步運行任務(wù)。
• 任務(wù)是獨立的，可以并行地運行。
• 工作容積變化很大，需要一個可擴展的解決方案。
• 該解決方案必須提供高可用性，并且如果處理一個任務(wù)失敗必須是有彈性的。

這種模式可能不適合時：

• 它是不容易的應(yīng)用程序的工作負荷分離成離散的任務(wù)，或有任務(wù)之間的依賴程度高。
• 任務(wù)必須同步進行，而應(yīng)用邏輯必須等待任務(wù)完成后再繼續(xù)。
• 任務(wù)必須以特定的順序來執(zhí)行。

Note

有些郵件系統(tǒng)支持會話，使生產(chǎn)者對消息進行分組在一起，并確保它們都被同一個接收者處理。這個機制可以與優(yōu)先消息使用（如果它們支持）來實現(xiàn)消息排序的一種形式，在順序從生產(chǎn)者傳送消息到單個消費者。

例子

Azure 提供存儲隊列和服務(wù)總線隊列，可作為一個合適的機制來實現(xiàn)這種模式。應(yīng)用邏輯可以發(fā)布消息到一個隊列，而消費者實現(xiàn)為在一個或多個角色的任務(wù)可以檢索從這個隊列中的消息并進行處理。對于彈性，一個服務(wù)總線隊列使得消費者使用 PeekLock 模式，當(dāng)它從隊列檢索消息。這種模式實際上不是刪除消息，而只是從其他消費者隱藏它。當(dāng)處理完它原來的用戶可以刪除該郵件。如果消費者要失敗，偷看鎖將超時，消息將再次變得可見，讓消費者又找回它。

Note

有關(guān)使用 Azure 的服務(wù)總線隊列的詳細信息，請參閱服務(wù)總線隊列，主題和 MSDN 上的訂閱。有關(guān)使用 Azure 存儲隊列的信息，請參閱如何 MSDN 上使用隊列存儲服務(wù)。

從可供下載的例子 CompetingConsumers 解決方案的 QueueManager 類下面的代碼顯示了本指南說明了如何通過在網(wǎng)絡(luò)或輔助角色開始的事件處理程序使用 QueueClient 實例中創(chuàng)建一個隊列。

private string queueName = ...;  
private string connectionString = ...;  
...  
?
public async Task Start()  
{  
  // Check if the queue already exists.  
  var manager = NamespaceManager.CreateFromConnectionString(this.connectionString);  
  if (!manager.QueueExists(this.queueName))  
  {  
    var queueDescription = new QueueDescription(this.queueName);  
?
    // Set the maximum delivery count for messages in the queue. A message   
    // is automatically dead-lettered after this number of deliveries. The  
    // default value for dead letter count is 10.  
    queueDescription.MaxDeliveryCount = 3;  
?
    await manager.CreateQueueAsync(queueDescription);  
  }  
  ...  
?
  // Create the queue client. By default the PeekLock method is used.  
  this.client = QueueClient.CreateFromConnectionString(  
    this.connectionString, this.queueName);  
}

下面的代碼片段顯示了一個應(yīng)用程序如何創(chuàng)建和發(fā)送一批消息隊列。

public async Task SendMessagesAsync()  
{  
  // Simulate sending a batch of messages to the queue.  
  var messages = new List<BrokeredMessage>();  
?
  for (int i = 0; i < 10; i++)  
  {  
    var message = new BrokeredMessage() { MessageId = Guid.NewGuid().ToString() };  
    messages.Add(message);  
  }  
  await this.client.SendBatchAsync(messages);  
}

下面的代碼顯示了如何消費服務(wù)實例可以從隊列中下一個事件驅(qū)動的方式接收消息。該 processMessageTask 參數(shù)的 ReceiveMessages 法為代表，它引用在收到消息時運行的代碼。此代碼是異步運行。

private ManualResetEvent pauseProcessingEvent;  
...  
?
public void ReceiveMessages(Func<BrokeredMessage, Task> processMessageTask)  
{  
  // Set up the options for the message pump.  
  var options = new OnMessageOptions();  
?
  // When AutoComplete is disabled it is necessary to manually  
  // complete or abandon the messages and handle any errors.  
  options.AutoComplete = false;  
  options.MaxConcurrentCalls = 10;  
  options.ExceptionReceived += this.OptionsOnExceptionReceived;  
?
  // Use of the Service Bus OnMessage message pump.   
  // The OnMessage method must be called once, otherwise an exception will occur.  
  this.client.OnMessageAsync(  
    async (msg) =>  
    {  
      // Will block the current thread if Stop is called.  
      this.pauseProcessingEvent.WaitOne();  
?
      // Execute processing task here.  
      await processMessageTask(msg);  
    },  
    options);  
}  
...  
?
private void OptionsOnExceptionReceived(object sender,   
  ExceptionReceivedEventArgs exceptionReceivedEventArgs)  
{  
  ...  
}

需要注意的是自動縮放的功能，例如可在天青，可用于啟動和停止的角色實例的隊列長度的波動。欲了解更多信息，請參閱自動縮放指導(dǎo)。另外，沒有必要維持角色實例和工人之間的一對一的對應(yīng)過程，單個角色實例可以實現(xiàn)多個工作進程。欲了解更多信息，請參閱計算資源整合模式。