# 云計算設計模式（十五）——管道和過濾器模式 分解，執行復雜處理成一系列可重復使用分立元件的一個任務。這種模式可以允許執行的處理進行部署和獨立縮放任務元素提高性能，可擴展性和可重用性。 ## 背景和問題 一個應用程序可能需要執行各種關于它處理的信息不同復雜的任務。一個簡單，但不靈活的方式來實施這個應用程序可以執行此處理為單一模塊。然而，這種方法有可能減少用于重構代碼，對其進行優化，或者重新使用它，如果是在應用程序中其他地方所需要的相同的處理的部件的機會。 圖1通過使用單片式的方式示出了與處理數據的問題。一個應用程序接收并處理來自兩個來源的數據進行處理。從每個源數據是由執行一系列任務來轉換該數據，并傳遞結果給應用程序的業務邏輯之前的獨立模塊進行處理。  圖1 - 使用單一模塊實現的解決方案 部分的單片模塊執行的任務在功能上是非常相似的，但在模塊已被分開設計的。實現該任務的代碼被緊密模塊內耦合，并且此代碼已開發具有很少或沒有給定重新使用或可伸縮性的思想。 然而，由每個模塊或每個任務的部署要求執行的處理任務，可能會改變，因為業務需求進行修改。有些任務可能是計算密集型的，并可能受益于強大的硬件上運行，而其他人可能并不需要如此昂貴的資源。此外，額外的處理可能需要在將來，或順序，其中由所述處理執行的任務可能會改變。一個解決方案是必需的，解決了這些問題，并且增加的可能性代碼重用。 ## 解決方案 分解需要為每個數據流轉換為一組離散的元件（或過濾器）的處理，其中每一個執行單任務。通過標準化每個組件接收和發射的數據的格式，這些過濾器可以組合在一起成為一個管道。這有助于避免重復代碼，并且可以很容易地移除，替換或集成額外的組件，如果處理要求改變。圖2顯示了這種結構的一個例子。  圖2 - 通過使用管道和過濾器實現的解決方案 處理一個請求所花費的時間取決于最慢的過濾器管道中的速度。這可能是一個或多個濾波器可能被證明是一個瓶頸，尤其是如果出現在從一個特定的數據源的數據流的大量請求。流水線結構的一個關鍵優點是它提供了機會，運行速度慢的過濾器的并聯情況下，使系統能夠分散負載并提高吞吐量。 可以獨立縮放組成一個管道可以在不同的機器上運行過濾器，使他們和可以利用的彈性，許多云計算環境提供的優勢。過濾器是計算密集型可以在高性能的硬件上運行，而其他要求不高的過濾器可以對商品（便宜）的硬件來承載。過濾器甚至不需要是在同一數據中心或地理位置，它允許在一個管道中的每個元素的環境下接近它需要的資源來運行。 圖3示出了從源 1 施加到管道中的數據的一個例子。  圖3 - 在一個管道負載平衡組件 如果一個濾波器的輸入和輸出被構造為一個流，它可能是能夠進行的處理并行的每個過濾器。在流水線的第一個過濾器可以開始工作，并開始發射其結果，它們會直接傳遞到序列中的下一個過濾器之前的第一過濾器已經完成它的工作。 另一個好處是靈活性，這種模式可以提供。如果一個過濾器發生故障或者其上運行的機器不再可用時，管道可能能夠重新安排濾波器所執行的工作，并指示此工作到組件的另一個實例。單個過濾器的故障不會必然導致整個管道的故障。 使用管道和過濾器與補償交易模式相結合的模式可以提供一種替代的方法來實現分布式事務。分布式事務可以被分解成單獨的賠的任務，每個都可以通過使用一個過濾器，也實現了補償事務圖案來實現。在一個管道中的過濾器可以在運行接近它們保持數據被實現為單獨的托管工作。 ## 問題和注意事項 在決定如何實現這個模式時，您應考慮以下幾點： - 復雜性。增加的靈活性，這種模式提供了還可以引入復雜性，特別是如果被分布在不同的服務器上在管道的過濾器。 - 可靠性。使用一個基礎結構，可以確保在管道中的過濾器之間流動的數據也不會丟失。 - 冪等性。如果在管道中的過濾失敗接收到消息后，任務被重新調度到過濾器的另一個實例，所述部分工作可能已經完成。如果這個工作更新的全局狀態的某些方面（如存儲在數據庫中的信息），同樣更新可以重復。如果公布的結果，在管道中的下一個過濾器后，過濾器出現故障，但在此之前表示，該公司已經成功地完成了它的工作可能會出現類似的問題。在這些情況下，相同的工作可以由過濾器的另一個實例被重復，導致相同的結果要貼兩次。這可能導致在管道隨后過濾兩次處理相同的數據。因此，在一個管道的過濾器應該被設計為冪等。欲了解更多信息，請參見喬納森·奧利弗的博客冪等模式。 - 重復的消息。如果在管道中的過濾器可以發布一個消息給流水線的下一個階段之后發生故障時，過濾器的另一個實例，可以執行（由冪等考慮以上所描述的），并且將發布相同消息的拷貝到流水線。這可能導致同樣的信息的兩個實例被傳遞到下一個過濾器。為了避免這種情況，該管道應檢測并消除重復的消息。 > 注意： 如果要實現管道使用消息隊列（如微軟的Azure服務總線隊列），消息隊列基礎設施可以提供自動重復消息檢測和清除。 - 上下文和狀態。在管道中，每個過濾器主要運行在孤立和不應該做這件事是如何被調用的任何假設。這意味著，每一個過濾器必須具有足夠的上下文與它能夠執行它的工作提供。這種情況下可包含相當數量的狀態信息。 ## 何時使用這個模式 使用這種模式時： - 由一個應用程序所需的處理可以很容易地被分解成一組離散的，獨立的步驟。 - 由應用程序執行的處理步驟具有不同的可擴展性要求。 注意：它可能會向組過濾器應擴展一起在相同的過程。欲了解更多信息，請參閱計算資源整合模式。 - 靈活性是必需的，以允許通過一個應用程序，或能力進行添加和刪除步驟中的處理步驟重新排序。 - 該系統可以受益于分配處理跨不同服務器的步驟。 - 一個可靠的解決方案是必需的，當數據正在被處理的最小化在一個步驟失敗的影響。 這種模式可能不適合時： - 通過應用程序執行的處理步驟并不是獨立的，或者他們必須共同作為同一事務的一部分來執行。 - 在一個步驟所需的上下文或狀態的信息量使得這種方法效率很低。它可能會持續狀態信息到數據庫代替，但不要使用此策略，如果在數據庫上的額外負載會導致過度競爭。 ## 例子 可以使用消息隊列的一個序列，以提供執行流水線所需的基礎設施。最初的消息隊列接收未處理的消息。實現為過濾器的任務偵聽此隊列的消息的組件，它執行其工作，然后投遞轉化的消息序列中的下一個隊列。另一個過濾器的任務可以偵聽在這個隊列中的消息，對其進行處理，后的結果到另一個隊列，依此類推，直到完全轉化的數據出現在隊列中的最后一個消息。  如果你正在構建一個解決方案，在 Azure 上，你可以使用服務總線隊列提供了可靠的，可擴展的排隊機制。下面所示的 ServiceBusPipeFilter 類提供了一個例子。它演示了如何實現接收從隊列中輸入消息，處理這些郵件的過濾器，并張貼結果到另一個隊列。 > 注意： 該 ServiceBusPipeFilter 類在 PipesAndFilters 解決方案 PipesAndFilters.Shared 項目定義。此示例代碼都可以可以下載本指導意見。 ~~~ public class ServiceBusPipeFilter { ... private readonly string inQueuePath; private readonly string outQueuePath; ... private QueueClient inQueue; private QueueClient outQueue; ... ? public ServiceBusPipeFilter(..., string inQueuePath, string outQueuePath = null) { ... this.inQueuePath = inQueuePath; this.outQueuePath = outQueuePath; } ? public void Start() { ... // Create the outbound filter queue if it does not exist. ... this.outQueue = QueueClient.CreateFromConnectionString(...); ? ... // Create the inbound and outbound queue clients. this.inQueue = QueueClient.CreateFromConnectionString(...); } ? public void OnPipeFilterMessageAsync( Func<BrokeredMessage, Task<BrokeredMessage>> asyncFilterTask, ...) { ... ? this.inQueue.OnMessageAsync( async (msg) => { ... // Process the filter and send the output to the // next queue in the pipeline. var outMessage = await asyncFilterTask(msg); ? // Send the message from the filter processor // to the next queue in the pipeline. if (outQueue != null) { await outQueue.SendAsync(outMessage); } ? // Note: There is a chance that the same message could be sent twice // or that a message may be processed by an upstream or downstream // filter at the same time. // This would happen in a situation where processing of a message was // completed, it was sent to the next pipe/queue, and then failed // to complete when using the PeekLock method. // Idempotent message processing and concurrency should be considered // in a real-world implementation. }, options); } ? public async Task Close(TimeSpan timespan) { // Pause the processing threads. this.pauseProcessingEvent.Reset(); ? // There is no clean approach for waiting for the threads to complete // the processing. This example simply stops any new processing, waits // for the existing thread to complete, then closes the message pump // and finally returns. Thread.Sleep(timespan); ? this.inQueue.Close(); ... } ? ... } ~~~ 在 ServiceBusPipeFilter 類 Start 方法連接到一對輸入和輸出隊列，以及關閉方法從輸入隊列斷開。該 OnPipeFilterMessageAsync 方法執行消息的實際處理;該 asyncFilterTask 參數這種方法指定要執行的處理。該 OnPipeFilterMessageAsync 方法等待輸入隊列中收到的消息，因為它到達，并張貼結果到輸出隊列通過運行在每個郵件的 asyncFilterTask 參數指定的代碼。隊列本身的構造函數中指定。 樣品溶液的過濾器實現了在一組工作角色。每個工人的作用可獨立進行調整，這取決于它執行的業務處理的復雜性，或者它需要執行此處理的資源。此外，各輔助角色的多個實例可以并行地運行，以提高吞吐量。 下面的代碼顯示了一個名為 PipeFilterARoleEntry 的 Azure 工作者角色，這是在樣品溶液中 PipeFilterA 項目定義。 ~~~ public class PipeFilterARoleEntry : RoleEntryPoint { ... private ServiceBusPipeFilter pipeFilterA; ? public override bool OnStart() { ... this.pipeFilterA = new ServiceBusPipeFilter( ..., Constants.QueueAPath, Constants.QueueBPath); ? this.pipeFilterA.Start(); ... } ? public override void Run() { this.pipeFilterA.OnPipeFilterMessageAsync(async (msg) => { // Clone the message and update it. // Properties set by the broker (Deliver count, enqueue time, ...) // are not cloned and must be copied over if required. var newMsg = msg.Clone(); ? await Task.Delay(500); // DOING WORK ? Trace.TraceInformation("Filter A processed message:{0} at {1}", msg.MessageId, DateTime.UtcNow); ? newMsg.Properties.Add(Constants.FilterAMessageKey, "Complete"); ? return newMsg; }); ? ... } ? ... } ~~~ 這個角色包含 ServiceBusPipeFilter 對象。在角色 OnStart 方法連接到隊列接收輸入的信息并張貼輸出消息（隊列的名稱在常量類中定義）。 Run 方法調用 OnPipeFilterMessagesAsync 方法來對接收到的（在本例中，該處理通過等待較短的時間段模擬的）的每個消息執行某些處理。何時處理完成時，一個新的消息被構造包含結果（在這種情況下，輸入消息被簡單地增加了一個自定義屬性），并將該消息發送到輸出隊列。 示例代碼中包含一個名為 PipeFilterBRoleEntry 在 PipeFilterB 項目的另一名工人的作用。這個角色類似于 PipeFilterARoleEntry 不同之處在于它的 Run 方法進行不同的處理。在本例中的解決方案，這兩種作用結合起來，構建一個管道;為 PipeFilterARoleEntry 角色輸出隊列是用于 PipeFilterBRoleEntry 角色的輸入隊列。 樣品溶液還提供了兩個名為 InitialSenderRoleEntry（在 InitialSender 項目）和 FinalReceiverRoleEntry（在 FinalReceiver 項目），進一步的角色。該 InitialSenderRoleEntry 作用提供了在管道中的初始消息。OnStart 方法連接到單個隊列和運行方法的帖子的方法來此隊列。這個隊列是所使用的 PipeFilterARoleEntry 作用，所以發布一條消息到這個隊列的輸入隊列導致由 PipeFilterARoleEntry 作用來接收和處理消息。經處理的信息，然后通過 PipeFilterBRoleEntry 作用傳遞。 為 FinalReceiveRoleEntry 角色輸入隊列是用于 PipeFilterBRoleEntry 角色的輸出隊列。 Run 方法在 FinalReceiveRoleEntry 作用，如下圖所示，接收到該消息，并且執行一些最后的處理。然后將其寫入了過濾器的管道跟蹤輸出添加自定義屬性的值。 ~~~ public class FinalReceiverRoleEntry : RoleEntryPoint { ... // Final queue/pipe in the pipeline from which to process data. private ServiceBusPipeFilter queueFinal; ? public override bool OnStart() { ... // Set up the queue. this.queueFinal = new ServiceBusPipeFilter(...,Constants.QueueFinalPath); this.queueFinal.Start(); ... } ? public override void Run() { this.queueFinal.OnPipeFilterMessageAsync( async (msg) => { await Task.Delay(500); // DOING WORK ? // The pipeline message was received. Trace.TraceInformation( "Pipeline Message Complete - FilterA:{0} FilterB:{1}", msg.Properties[Constants.FilterAMessageKey], msg.Properties[Constants.FilterBMessageKey]); ? return null; }); ... } ? ... } ~~~