在前面一篇文章中，我們看到了一個通用的作業處理框架，以及一些簡單的作業佇列處理的程式。作業佇列非常適合將一個輸入轉化為一個輸出，但是，有許多批處理應用程式需要執行多個操作，或者需要將單個數據輸入生成為多種不同的輸出。在這種情況下，我們開始將作業佇列連線在一起，以便使一個作業佇列的輸出成為一個或多個其他作業佇列的輸入，以此類推，這樣就形成了一系列事件響應的處理步驟。

這種事件驅動的處理系統通常被稱作為作業流系統，因為有一個作業流通過一個定向的非迴圈圖來描述這些作業流的各個階段和之間的關係，如圖1所示。

這種型別的系統最直接的應用就是將一個佇列的輸出連線到下一個佇列的輸入，但是隨著系統變得越來越複雜，出現了一系列不同的模式將作業佇列連線在一起，理解和設計這些模式對於理解系統是如何工作的非常重要。這種事件驅動的批處理的操作與事件驅動的FaaS很相似，因此，在不瞭解這些佇列是如何關聯在一起的情況下，是很難完全理解系統的執行方式的。

圖1 該作業流將複製的工作分為多個佇列（階段2a、2b）來並行處理這些佇列，並最後將結果彙總成單個佇列（階段3）

事件驅動處理的模式

除了前一篇文章介紹的簡單作業佇列之外，還有許多可以將作業佇列連線在一起的模式。其中最簡單的模式就是，單個佇列的輸出作為第二個佇列的輸入，在這裡我們不會介紹它，我們將介紹涉及協調多個不同的佇列，或者修改一個或多個作業佇列輸出的模式。

Copier

協調多個作業佇列的第一種模式是Copier模式。Copier的任務是將單個作業項流複製到多個相同的流當中，當對同一個作業項有不同的工作要做時，這種模式很有用。例如視訊的渲染，當渲染視訊時，根據視訊的顯示位置可以使用多種不同的格式，可能有4KB的高解析度格式用於硬體播放、1080畫素的資料流渲染、用於網速較低的移動使用者的低解析度，以及用於顯示部分電影情節的GIF動畫略縮圖等，所有這些作業項對於每個渲染都可以進行獨立建模，但是每個作業項的輸入都是相同的，圖2顯示了用於轉碼的Copier模式。

Filter

Filter是事件驅動處理的第二種模式。Filter的作用是，通過對作業流進行過濾，篩選出不符合特定標準的作業項，減少作業項流的數目。例如，設定一個用來處理新使用者註冊服務的批處理作業流，有些使用者會勾選複選框，表示他們希望商家通過電子郵件與他們聯絡以獲取促銷的資訊或者其他的資訊，在這樣一個作業流中，可以將新註冊使用者的集合過濾為僅顯示選擇進行聯絡的使用者。

理想情況下，可以組成一個過濾作業流的源來當作一個大使，用來包裝現有的作業佇列源，原始的源容器提供了要處理的作業項的完整列表，然後filter容器根據篩選條件來調整該列表，並將篩選結果返回給作業佇列的基礎架構，圖3展示了這種模式。

Splitter

有時候，對作業項進行過濾的時候，並不想把過濾出來的一部分作業丟掉，而是希望通過將作業佇列分為兩個單獨的作業佇列來形成兩種不同的輸入，對於這種任務，就需要使用Splitter。Splitter的作用是對作業的一些標準進行評估，就像Filter一樣，但並不是丟掉一些輸入，而是根據給定的標準向不同的佇列傳送不同的輸入。

Splitter模式的一個例子就是處理線上訂單，人們可以通過電子郵件或者簡訊的方式來接收快遞的通知。給定已經發貨的物品的作業佇列，Splitter將其分成兩個不同的佇列：一個負責傳送電子郵件，另外一個負責傳送簡訊。如果Splitter需要將相同的輸出傳送到多個佇列中，則它也可以是Copier，例如，在前面的那個例子中，使用者同時選擇了簡訊和郵件。有意思的是，Splitter實際上也可以由Copier和兩個不同的Slipper來實現，但是Splitter模式是一種更加簡潔的表現形式，如圖4所示是使用Splitter模式向用戶傳送快遞訊息。

Sharder

一個稍微更加通用一些的Splitter就是Sharder，就像我們在之前的文章中看到的分片服務一樣，Sharder的作用是，在工作流中根據某種分片函式將單個佇列分成平均分配的作業項集合。對作業流進行分片可能有以下幾點原因：首先是可靠性，如果發生了由於更新錯誤、基礎架構故障或者其他問題，導致了單個作業流失敗，通過對作業佇列進行分割，這樣僅僅會影響到部分服務而不會影響到所有的服務。

例如，假設你向worker容器推送了一個壞的更新，這會導致workers崩潰，並且會導致作業處理的停止，如果你只有一個作業佇列來對作業進行處理，那麼你的服務就會完全中斷並且影響到所有的使用者。相反，如果將作業佇列分成四個不同的分片，相當於所有的作業交給了四個不同的worker來進行處理，那麼假設有一個分片由於某種原因失敗了，意味著只有四分之一的使用者會受到影響。

對作業佇列進行分片的另外一個原因是為了更平均地分配不同資源的作業，如果你並不關心哪個區域或者資料中心用於處理一些特定的作業項集，則可以使用sharder將作業平均地分佈到多個數據中心，以平均所有資料中心／區域的資源利用率。並且，將作業佇列分佈到多個數據中心／區域，當有故障發生時也能更好的提供可靠性，圖5顯示了一切正常工作時的sharded佇列。

當一些shards由於故障失敗時，即使只剩下單個佇列，分片演算法也會進行動態調整以將作業傳送到其餘健康的作業佇列中，如圖6所示。

Merger

事件驅動批處理的最後一種模式就是Merger模式。Merger與Copier相反，它是使用兩個不同的作業佇列，並將他們變成一個單一的作業佇列。例如，假設你擁有大量不同的repositories，並且所有的repositories同時提交了新的commit請求，你想要處理每個commit並對其進行build和測試。對每個repositories建立單獨的基礎架構是很難的，我們可以將每個不同的repositories建模為獨立的作業佇列源，這些佇列提供一個commits的集合，我們可以使用一個merger介面卡，將所有不同的作業佇列的輸入轉換為一組合並輸入，這個合併的commits流是執行操作時的唯一來源。Merger是介面卡模式中一個很好的例子，雖然在這種情況下，介面卡實際上是將多個執行中的source容器合併到一個源中，這個多介面卡模式如圖7所示。

Publisher／Subscriber的基礎結構

我們已經看到各種將不同的事件驅動批處理模式連線在一起的抽象模式，但是當真正去構建這樣的系統時，我們需要弄清楚如何去管理資料流。最簡單的方法是，簡單地將作業佇列中的每個元素寫入本地檔案系統上的特定目錄，然後讓每個階段都去監視該目錄以進行輸入。但是，使用本地檔案系統會限制我們的作業流在單個節點上的執行，我們可以引入一個網路檔案系統來將檔案分發到多個節點，但是這樣會增加程式碼和批量作業流的複雜度。

相反，構建這種作業流的一種很流行的方法是使用Publisher／Subscriber（pub／sub）API或服務，pub／sub API允許使用者定義一組佇列（有時候稱其為topics），一個或多個Publishers向這些佇列釋出訊息。同樣，一個或多個subscribers監聽著這些佇列中的新訊息，當一個訊息釋出時，訊息會被可靠的儲存在佇列中，然後以可靠的方式傳送給Subscribers。

當下，大多數的公有云都具有pub／sub API，如Azure的EventGrid或Amazon的簡單佇列服務。此外，開源的Kafka專案提供了非常受歡迎的pub／sub實現，可以在自己的硬體和雲虛擬機器上執行。