（十五）—— 管道和過濾器模式

云計算設(shè)計模式（十五）——管道和過濾器模式

分解，執(zhí)行復(fù)雜處理成一系列可重復(fù)使用分立元件的一個任務(wù)。這種模式可以允許執(zhí)行的處理進(jìn)行部署和獨立縮放任務(wù)元素提高性能，可擴展性和可重用性。

背景和問題

一個應(yīng)用程序可能需要執(zhí)行各種關(guān)于它處理的信息不同復(fù)雜的任務(wù)。一個簡單，但不靈活的方式來實施這個應(yīng)用程序可以執(zhí)行此處理為單一模塊。然而，這種方法有可能減少用于重構(gòu)代碼，對其進(jìn)行優(yōu)化，或者重新使用它，如果是在應(yīng)用程序中其他地方所需要的相同的處理的部件的機會。

圖1通過使用單片式的方式示出了與處理數(shù)據(jù)的問題。一個應(yīng)用程序接收并處理來自兩個來源的數(shù)據(jù)進(jìn)行處理。從每個源數(shù)據(jù)是由執(zhí)行一系列任務(wù)來轉(zhuǎn)換該數(shù)據(jù)，并傳遞結(jié)果給應(yīng)用程序的業(yè)務(wù)邏輯之前的獨立模塊進(jìn)行處理。

圖1 - 使用單一模塊實現(xiàn)的解決方案

部分的單片模塊執(zhí)行的任務(wù)在功能上是非常相似的，但在模塊已被分開設(shè)計的。實現(xiàn)該任務(wù)的代碼被緊密模塊內(nèi)耦合，并且此代碼已開發(fā)具有很少或沒有給定重新使用或可伸縮性的思想。

然而，由每個模塊或每個任務(wù)的部署要求執(zhí)行的處理任務(wù)，可能會改變，因為業(yè)務(wù)需求進(jìn)行修改。有些任務(wù)可能是計算密集型的，并可能受益于強大的硬件上運行，而其他人可能并不需要如此昂貴的資源。此外，額外的處理可能需要在將來，或順序，其中由所述處理執(zhí)行的任務(wù)可能會改變。一個解決方案是必需的，解決了這些問題，并且增加的可能性代碼重用。

解決方案

分解需要為每個數(shù)據(jù)流轉(zhuǎn)換為一組離散的元件（或過濾器）的處理，其中每一個執(zhí)行單任務(wù)。通過標(biāo)準(zhǔn)化每個組件接收和發(fā)射的數(shù)據(jù)的格式，這些過濾器可以組合在一起成為一個管道。這有助于避免重復(fù)代碼，并且可以很容易地移除，替換或集成額外的組件，如果處理要求改變。圖2顯示了這種結(jié)構(gòu)的一個例子。

圖2 - 通過使用管道和過濾器實現(xiàn)的解決方案

處理一個請求所花費的時間取決于最慢的過濾器管道中的速度。這可能是一個或多個濾波器可能被證明是一個瓶頸，尤其是如果出現(xiàn)在從一個特定的數(shù)據(jù)源的數(shù)據(jù)流的大量請求。流水線結(jié)構(gòu)的一個關(guān)鍵優(yōu)點是它提供了機會，運行速度慢的過濾器的并聯(lián)情況下，使系統(tǒng)能夠分散負(fù)載并提高吞吐量。

可以獨立縮放組成一個管道可以在不同的機器上運行過濾器，使他們和可以利用的彈性，許多云計算環(huán)境提供的優(yōu)勢。過濾器是計算密集型可以在高性能的硬件上運行，而其他要求不高的過濾器可以對商品（便宜）的硬件來承載。過濾器甚至不需要是在同一數(shù)據(jù)中心或地理位置，它允許在一個管道中的每個元素的環(huán)境下接近它需要的資源來運行。

圖3示出了從源 1 施加到管道中的數(shù)據(jù)的一個例子。

圖3 - 在一個管道負(fù)載平衡組件

如果一個濾波器的輸入和輸出被構(gòu)造為一個流，它可能是能夠進(jìn)行的處理并行的每個過濾器。在流水線的第一個過濾器可以開始工作，并開始發(fā)射其結(jié)果，它們會直接傳遞到序列中的下一個過濾器之前的第一過濾器已經(jīng)完成它的工作。

另一個好處是靈活性，這種模式可以提供。如果一個過濾器發(fā)生故障或者其上運行的機器不再可用時，管道可能能夠重新安排濾波器所執(zhí)行的工作，并指示此工作到組件的另一個實例。單個過濾器的故障不會必然導(dǎo)致整個管道的故障。

使用管道和過濾器與補償交易模式相結(jié)合的模式可以提供一種替代的方法來實現(xiàn)分布式事務(wù)。分布式事務(wù)可以被分解成單獨的賠的任務(wù)，每個都可以通過使用一個過濾器，也實現(xiàn)了補償事務(wù)圖案來實現(xiàn)。在一個管道中的過濾器可以在運行接近它們保持?jǐn)?shù)據(jù)被實現(xiàn)為單獨的托管工作。

問題和注意事項

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

• 復(fù)雜性。增加的靈活性，這種模式提供了還可以引入復(fù)雜性，特別是如果被分布在不同的服務(wù)器上在管道的過濾器。
• 可靠性。使用一個基礎(chǔ)結(jié)構(gòu)，可以確保在管道中的過濾器之間流動的數(shù)據(jù)也不會丟失。
• 冪等性。如果在管道中的過濾失敗接收到消息后，任務(wù)被重新調(diào)度到過濾器的另一個實例，所述部分工作可能已經(jīng)完成。如果這個工作更新的全局狀態(tài)的某些方面（如存儲在數(shù)據(jù)庫中的信息），同樣更新可以重復(fù)。如果公布的結(jié)果，在管道中的下一個過濾器后，過濾器出現(xiàn)故障，但在此之前表示，該公司已經(jīng)成功地完成了它的工作可能會出現(xiàn)類似的問題。在這些情況下，相同的工作可以由過濾器的另一個實例被重復(fù)，導(dǎo)致相同的結(jié)果要貼兩次。這可能導(dǎo)致在管道隨后過濾兩次處理相同的數(shù)據(jù)。因此，在一個管道的過濾器應(yīng)該被設(shè)計為冪等。欲了解更多信息，請參見喬納森·奧利弗的博客冪等模式。
• 重復(fù)的消息。如果在管道中的過濾器可以發(fā)布一個消息給流水線的下一個階段之后發(fā)生故障時，過濾器的另一個實例，可以執(zhí)行（由冪等考慮以上所描述的），并且將發(fā)布相同消息的拷貝到流水線。這可能導(dǎo)致同樣的信息的兩個實例被傳遞到下一個過濾器。為了避免這種情況，該管道應(yīng)檢測并消除重復(fù)的消息。

注意： 如果要實現(xiàn)管道使用消息隊列（如微軟的Azure服務(wù)總線隊列），消息隊列基礎(chǔ)設(shè)施可以提供自動重復(fù)消息檢測和清除。

• 上下文和狀態(tài)。在管道中，每個過濾器主要運行在孤立和不應(yīng)該做這件事是如何被調(diào)用的任何假設(shè)。這意味著，每一個過濾器必須具有足夠的上下文與它能夠執(zhí)行它的工作提供。這種情況下可包含相當(dāng)數(shù)量的狀態(tài)信息。

何時使用這個模式

使用這種模式時：

• 由一個應(yīng)用程序所需的處理可以很容易地被分解成一組離散的，獨立的步驟。
• 由應(yīng)用程序執(zhí)行的處理步驟具有不同的可擴展性要求。

注意：它可能會向組過濾器應(yīng)擴展一起在相同的過程。欲了解更多信息，請參閱計算資源整合模式。

• 靈活性是必需的，以允許通過一個應(yīng)用程序，或能力進(jìn)行添加和刪除步驟中的處理步驟重新排序。
• 該系統(tǒng)可以受益于分配處理跨不同服務(wù)器的步驟。
• 一個可靠的解決方案是必需的，當(dāng)數(shù)據(jù)正在被處理的最小化在一個步驟失敗的影響。

這種模式可能不適合時：

• 通過應(yīng)用程序執(zhí)行的處理步驟并不是獨立的，或者他們必須共同作為同一事務(wù)的一部分來執(zhí)行。
• 在一個步驟所需的上下文或狀態(tài)的信息量使得這種方法效率很低。它可能會持續(xù)狀態(tài)信息到數(shù)據(jù)庫代替，但不要使用此策略，如果在數(shù)據(jù)庫上的額外負(fù)載會導(dǎo)致過度競爭。

例子

可以使用消息隊列的一個序列，以提供執(zhí)行流水線所需的基礎(chǔ)設(shè)施。最初的消息隊列接收未處理的消息。實現(xiàn)為過濾器的任務(wù)偵聽此隊列的消息的組件，它執(zhí)行其工作，然后投遞轉(zhuǎn)化的消息序列中的下一個隊列。另一個過濾器的任務(wù)可以偵聽在這個隊列中的消息，對其進(jìn)行處理，后的結(jié)果到另一個隊列，依此類推，直到完全轉(zhuǎn)化的數(shù)據(jù)出現(xiàn)在隊列中的最后一個消息。

如果你正在構(gòu)建一個解決方案，在 Azure 上，你可以使用服務(wù)總線隊列提供了可靠的，可擴展的排隊機制。下面所示的 ServiceBusPipeFilter 類提供了一個例子。它演示了如何實現(xiàn)接收從隊列中輸入消息，處理這些郵件的過濾器，并張貼結(jié)果到另一個隊列。

注意： 該 ServiceBusPipeFilter 類在 PipesAndFilters 解決方案 PipesAndFilters.Shared 項目定義。此示例代碼都可以可以下載本指導(dǎo)意見。

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 方法連接到一對輸入和輸出隊列，以及關(guān)閉方法從輸入隊列斷開。該 OnPipeFilterMessageAsync 方法執(zhí)行消息的實際處理;該 asyncFilterTask 參數(shù)這種方法指定要執(zhí)行的處理。該 OnPipeFilterMessageAsync 方法等待輸入隊列中收到的消息，因為它到達(dá)，并張貼結(jié)果到輸出隊列通過運行在每個郵件的 asyncFilterTask 參數(shù)指定的代碼。隊列本身的構(gòu)造函數(shù)中指定。

樣品溶液的過濾器實現(xiàn)了在一組工作角色。每個工人的作用可獨立進(jìn)行調(diào)整，這取決于它執(zhí)行的業(yè)務(wù)處理的復(fù)雜性，或者它需要執(zhí)行此處理的資源。此外，各輔助角色的多個實例可以并行地運行，以提高吞吐量。

下面的代碼顯示了一個名為 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 方法調(diào)用 OnPipeFilterMessagesAsync 方法來對接收到的（在本例中，該處理通過等待較短的時間段模擬的）的每個消息執(zhí)行某些處理。何時處理完成時，一個新的消息被構(gòu)造包含結(jié)果（在這種情況下，輸入消息被簡單地增加了一個自定義屬性），并將該消息發(fā)送到輸出隊列。

示例代碼中包含一個名為 PipeFilterBRoleEntry 在 PipeFilterB 項目的另一名工人的作用。這個角色類似于 PipeFilterARoleEntry 不同之處在于它的 Run 方法進(jìn)行不同的處理。在本例中的解決方案，這兩種作用結(jié)合起來，構(gòu)建一個管道;為 PipeFilterARoleEntry 角色輸出隊列是用于 PipeFilterBRoleEntry 角色的輸入隊列。

樣品溶液還提供了兩個名為 InitialSenderRoleEntry（在 InitialSender 項目）和 FinalReceiverRoleEntry（在 FinalReceiver 項目），進(jìn)一步的角色。該 InitialSenderRoleEntry 作用提供了在管道中的初始消息。OnStart 方法連接到單個隊列和運行方法的帖子的方法來此隊列。這個隊列是所使用的 PipeFilterARoleEntry 作用，所以發(fā)布一條消息到這個隊列的輸入隊列導(dǎo)致由 PipeFilterARoleEntry 作用來接收和處理消息。經(jīng)處理的信息，然后通過 PipeFilterBRoleEntry 作用傳遞。

為 FinalReceiveRoleEntry 角色輸入隊列是用于 PipeFilterBRoleEntry 角色的輸出隊列。 Run 方法在 FinalReceiveRoleEntry 作用，如下圖所示，接收到該消息，并且執(zhí)行一些最后的處理。然后將其寫入了過濾器的管道跟蹤輸出添加自定義屬性的值。

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