（四）——消費(fèi)者的競(jìng)爭(zhēng)模式

云計(jì)算設(shè)計(jì)模式（四）——消費(fèi)者的競(jìng)爭(zhēng)模式

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

背景和問(wèn)題

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

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

解決方案

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

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

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

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

問(wèn)題和注意事項(xiàng)

在決定如何實(shí)現(xiàn)這個(gè)模式時(shí)，請(qǐng)考慮以下幾點(diǎn)：

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

注意

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

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

注意

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

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

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

使用這種模式時(shí)：

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

這種模式可能不適合時(shí)：

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

Note

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

例子

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

Note

有關(guān)使用 Azure 的服務(wù)總線(xiàn)隊(duì)列的詳細(xì)信息，請(qǐng)參閱服務(wù)總線(xiàn)隊(duì)列，主題和 MSDN 上的訂閱。有關(guān)使用 Azure 存儲(chǔ)隊(duì)列的信息，請(qǐng)參閱如何 MSDN 上使用隊(duì)列存儲(chǔ)服務(wù)。

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

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);  
}

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

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);  
}

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

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òng)縮放的功能，例如可在天青，可用于啟動(dòng)和停止的角色實(shí)例的隊(duì)列長(zhǎng)度的波動(dòng)。欲了解更多信息，請(qǐng)參閱自動(dòng)縮放指導(dǎo)。另外，沒(méi)有必要維持角色實(shí)例和工人之間的一對(duì)一的對(duì)應(yīng)過(guò)程，單個(gè)角色實(shí)例可以實(shí)現(xiàn)多個(gè)工作進(jìn)程。欲了解更多信息，請(qǐng)參閱計(jì)算資源整合模式。