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

背景和問題

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

請求的數(shù)量可以隨著時(shí)間的原因有很多顯著變化。突然一陣在用戶活動或聚集的請求，來自多個(gè)租戶未來可能會導(dǎo)致不可預(yù)測的工作負(fù)載。在高峰時(shí)間的系統(tǒng)可能需要處理許多每秒數(shù)百個(gè)請求，而在其他時(shí)間的數(shù)量可能是非常小的。此外，該工作的性質(zhì)進(jìn)行的處理這些請求可能是高度可變的。使用消費(fèi)者服務(wù)的單個(gè)實(shí)例，可能會導(dǎo)致該實(shí)例成為充斥請求或消息傳送系統(tǒng)可通過消息從應(yī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ì)列來實(shí)現(xiàn)應(yīng)用和消費(fèi)者服務(wù)的實(shí)例之間的通信信道。在消息隊(duì)列中的形式應(yīng)用帖請求，以及消費(fèi)者的服務(wù)實(shí)例從隊(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ā)送請求量很大的變化。隊(duì)列充當(dāng)應(yīng)用程序?qū)嵗拖M(fèi)者服務(wù)實(shí)例，這有助于最大限度地減少對應(yīng)用程序和服務(wù)實(shí)例（所描述的基于隊(duì)列的負(fù)載調(diào)平模式）的可用性和響應(yīng)性的影響之間的緩沖區(qū)。處理的消息，需要一些將被執(zhí)行時(shí)，不會妨礙同時(shí)由消費(fèi)者服務(wù)的其他實(shí)例所處理的其它消息長期運(yùn)行的處理。

• 它提高了可靠性。如果一個(gè)生產(chǎn)者直接與消費(fèi)者，而不是使用這種模式進(jìn)行通信，但不監(jiān)視消費(fèi)者，有一個(gè)高概率的消息可能丟失或失敗，如果消費(fèi)者無法進(jìn)行處理。在這種模式的消息不被發(fā)送給一個(gè)特定的服務(wù)實(shí)例，一個(gè)失敗服務(wù)實(shí)例不會阻塞一個(gè)生產(chǎn)者和消息可以通過任何加工服務(wù)實(shí)例進(jìn)行處理。

• 它不需要復(fù)雜的協(xié)調(diào)的消費(fèi)者之間，或在生產(chǎn)者和消費(fèi)者的實(shí)例。消息隊(duì)列確保每個(gè)消息傳遞至少一次。

• 它是可擴(kuò)展的。該系統(tǒng)能夠動態(tài)地增加或減少消費(fèi)者服務(wù)的實(shí)例的數(shù)目的消息量是波動的。

• 它可以提高彈性，如果消息隊(duì)列提供事務(wù)讀取操作。如果消費(fèi)者服務(wù)實(shí)例能夠讀取和處理該消息作為一個(gè)事務(wù)操作的一部分，并且如果這種消費(fèi)服務(wù)實(shí)例隨后發(fā)生故障時(shí)，這種模式可以確保該消息將被返回到隊(duì)列中被拾起并處理通過的另一個(gè)實(shí)例消費(fèi)者服務(wù)。

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

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

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

注意

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

• 設(shè)計(jì)服務(wù)的永續(xù)性。如果系統(tǒng)被設(shè)計(jì)為檢測和重新啟動失敗的服務(wù)實(shí)例中，可能有必要執(zhí)行由服務(wù)實(shí)例執(zhí)行作為冪等操作，以最小化被檢索和處理一次以上的單個(gè)消息的影響的處理。

• 檢測有害消息。格式不正確的消息，或者需要訪問不可用的資源的任務(wù)，可能會造成服務(wù)實(shí)例失敗。該系統(tǒng)應(yīng)避免這樣的消息被返回到隊(duì)列，而是捕獲和別處存儲這些信息的詳細(xì)信息，以便可以在需要進(jìn)行分析。

• 處理結(jié)果。服務(wù)實(shí)例處理一個(gè)消息從生成該消息的應(yīng)用程序邏輯完全分離，并且它們未必能夠直接進(jìn)行通信。如果服務(wù)實(shí)例生成必須傳回給應(yīng)用程序邏輯結(jié)果，該信息必須被存儲在一個(gè)位置，都可以訪問兩個(gè)和系統(tǒng)必須提供某種指示時(shí)的處理已經(jīng)完成，以防止應(yīng)用邏輯從檢索數(shù)據(jù)不全。

注意

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

• 擴(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ì)列的消息，它也不會丟失。這是確保所有郵件傳遞至少一次重要的。

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

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

• 工作量為一個(gè)應(yīng)用程序被分成可異步運(yùn)行任務(wù)。

• 任務(wù)是獨(dú)立的，可以并行地運(yùn)行。

• 工作容積變化很大，需要一個(gè)可擴(kuò)展的解決方案。

• 該解決方案必須提供高可用性，并且如果處理一個(gè)任務(wù)失敗必須是有彈性的。

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

• 它是不容易的應(yīng)用程序的工作負(fù)荷分離成離散的任務(wù)，或有任務(wù)之間的依賴程度高。

• 任務(wù)必須同步進(jìn)行，而應(yīng)用邏輯必須等待任務(wù)完成后再繼續(xù)。

• 任務(wù)必須以特定的順序來執(zhí)行。

Note

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

例子

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

Note

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

從可供下載的例子 CompetingConsumers 解決方案的 QueueManager 類下面的代碼顯示了本指南說明了如何通過在網(wǎng)絡(luò)或輔助角色開始的事件處理程序使用 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ū)動的方式接收消息。該 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) ? { ?... ? } ?

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

相關(guān)文章:?

• 云計(jì)算設(shè)計(jì)模式（一）緩存預(yù)留模式
  

• 云計(jì)算設(shè)計(jì)模式（二）——斷路器模式
  

• 云計(jì)算設(shè)計(jì)模式（三）——補(bǔ)償交易模式
  

原文地址：http://blog.csdn.net/yangzhenping/article/details/40787219

---

.NET社區(qū)新聞，深度好文，微信中搜索dotNET跨平臺或掃描二維碼關(guān)注

總結(jié)

以上是生活随笔為你收集整理的云计算设计模式（四）——消费者的竞争模式的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。