生產者 vs 消費者 - BlockQueue 實作

摘要提示

• 生產者/消費者模式: 將工作分為前後兩階段，由生產者產生資料、消費者處理資料，以佇列協調進度與資源。
• 問題動機: 一般 ThreadPool 易用但不適合階段式流程，需以模式設計來平衡前後階段的吞吐。
• 實務案例: 下載檔案與壓縮 ZIP 同步進行，分別為 I/O bound 與 CPU bound，適合並行。
• 佇列為中介: 以佇列作為生產者/消費者之間的緩衝，並以容量限制避免資源暴衝。
• BlockQueue 需求: 需有容量上限、空佇列/滿佇列時會阻塞、支援關機（Shutdown）。
• 使用範例: 多個 Producer/Consumer 執行緒透過 BlockQueue 傳遞字串，完成後呼叫 Shutdown 結束。
• 同步機制: 以 ManualResetEvent 實作 Enqueue/Dequeue 的阻塞與喚醒，避免例外流控制。
• 成效觀察: 調整執行緒數可見供需平衡；高產出會受限於容量，高消費會即時取走工作。
• 關機語意: Shutdown 後不再允許 Enqueue，剩餘項目可被取盡，空時再依原生 Queue 行為拋例外。
• 後續延伸: 可改用環狀佇列或優先佇列，並可推展到 Pipeline 與 Stream 應用。

全文重點

本文從多執行緒「工具」轉向「模式」的實務設計，聚焦生產者/消費者（Producer/Consumer）。當工作流程可拆成兩個各自可並行、且資源型態不同的階段（如 I/O bound 的下載與 CPU bound 的壓縮），若直接用 ThreadPool 撒工作並不適配，反而需要一個中介緩衝來協調前後段的速率與資源壓力。作者提出以佇列（Queue）為中介的設計，讓生產者將成果放入佇列，消費者從佇列取出處理，並強調「可用性」與「封裝性」，使其能像泛型集合一樣被重複使用，而非課堂習作。

針對標準 Queue 不具備的行為，作者自製 BlockQueue：具容量上限，當滿則 Enqueue 阻塞、當空則 Dequeue 阻塞，並提供 Shutdown，以免無限等待。使用範例中，主程式建立多個 Producer/Consumer 執行緒，Producer 產生固定數量的字串放入佇列，全部完成後呼叫 Shutdown；Consumer 迴圈取出資料處理，遇到關機語意後在例外中結束。透過調整執行緒數量可觀察到：生產快時，佇列內的「庫存」受容量限制而不會無限膨脹；消費快時，幾乎即時清空，呈現供不應求。

BlockQueue 的核心是用 ManualResetEvent 搭配 lock 控制兩端阻塞/喚醒：Enqueue 在滿佇列時等待，Dequeue 在空佇列時等待；當對方放入或取出時設定對應事件，達到進度平衡。Shutdown 後不再接受新的 Enqueue，Dequeue 仍可取至清空，之後依原生 Queue 行為拋例外，讓消費端得以結束。最後，作者指出可進一步以環狀佇列最佳化固定大小的緩衝，或以 Priority Queue 實作插隊策略，並暗示此模式可延伸到多階段 Pipeline 與串流應用，將成為實作高效併行處理的基礎。

段落重點

從執行緒工具到程式設計模式的轉換

作者回顧既有多執行緒文章，多在講解如何精準控制執行緒本身；然而真正的效益取決於如何安排程式結構。當工作無法單純拆成許多獨立小任務交給 ThreadPool 時，就該考慮更高層級的模式。本文選擇「生產者/消費者」作為焦點，強調這是教科書級的經典課題，但目標是用 C#/.NET 實作出「可重用且易用」的解決方案，而非僅止於學術範例。

生產者/消費者模式與現實場景

以「大量下載並同時壓縮成 ZIP」為例：下載為 I/O bound，壓縮為 CPU bound，兩者資源競爭低，理應併行最佳。問題在於兩階段彼此卡資料依賴，不易直接獨立執行。引入「兩組人」比喻：下載完成即交給壓縮，不必等待全部下載。若暫存空間有限，需能在滿時暫停下載、在空時暫停壓縮，等資源回到平衡再繼續，這正是生產者/消費者需解決的供需調節。

以佇列作為中介與封裝的重要性

作者主張好的實作需像 System.Collections.Generic 對資料結構那樣，將細節封裝、接口直觀、易於重用。雖然可以用 Semaphore 等機制達成，但更推薦以 Queue 作為中介，因其語意自然貼近「庫存」。然而標準 Queue 不會回報滿載，也在空佇列時以例外回應，不符實務需求。因此需要一個「會阻塞而非拋例外」的 Queue，並且支援關機語意，避免消費者無限等待。

BlockQueue 的目標與規格

BlockQueue 預期有三點差異：可設定容量上限；當滿載時 Enqueue 阻塞、當為空時 Dequeue 阻塞；提供 Shutdown 可通知「不再補貨」，使消費者得以在清空後結束。如此即可將生產與消費以固定緩衝隔離，平衡兩端速度，並能在任務完成時優雅收尾。核心理念是讓使用方式盡量貼近一般 Queue，但加上阻塞與關機行為。

使用範例：多 Producer/Consumer 的主程式

示範程式建立一個容量 10 的 BlockQueue，開多個 Producer 與 Consumer 執行緒。Producer 每次隨機等待後產生字串並 Enqueue，重複固定次數，完成後寫出訊息。主執行緒等待所有 Producer 完成後呼叫 queue.Shutdown()，再等待所有 Consumer 結束。Consumer 以無限迴圈 Dequeue 取得資料並輸出，當 Dequeue 在關機後最終拋例外時跳出並結束。範例重點在展現使用介面直覺與收斂行為清晰。

觀察不同供需配置下的行為

當 Producer 較多（如 P:10/C:5）時，生產者領先消費者約 10 筆，之後因佇列容量受限而自動暫停，證明庫存不會無限膨脹；當 Consumer 較多（如 P:5/C:10）時，呈現供不應求，生產的項目幾乎立即被取走。這驗證了 BlockQueue 能有效穩定緩衝的大小，並根據供需變化自動調節兩端節奏。

BlockQueue 的同步實作與關機語意

BlockQueue 內部以 Queue 搭配兩個 ManualResetEvent（_enqueue_wait、_dequeue_wait）與 lock 實作：EnQueue 在偵測到未滿時加入項目，重設/設定對應事件，否則 WaitOne 阻塞；DeQueue 在非關機時檢查是否有項目，有則取出並調整事件，否則等待；在 Shutdown 後，Enqueue 被禁止（拋出例外），Dequeue 允許將庫存取盡，空佇列時恢復原生 Queue 的例外，讓消費者得以結束迴圈，完成流程的自然收束。

可能的優化與延伸應用

作者指出固定大小緩衝更適合用環狀佇列（Circular Queue）以提升效率；亦可導入 Priority Queue 實現插隊，讓高優先級任務先被處理，進一步強化生產者/消費者模式的調度能力。更進一步可擴展到多階段的 Pipeline（生產線）或串流型處理，形成連續加工流程。文末預告將在後續文章更完整探討此類模式與應用設計。

資訊整理

知識架構圖

1. 前置知識：
   • 作業系統中的 Producer-Consumer（有界緩衝區）問題
   • 多執行緒基礎：Thread、ThreadPool、鎖（lock）、競態條件
   • .NET 同步原語：ManualResetEvent、Semaphore（概念比較）
   • 基本資料結構：Queue、Circular Queue（環形佇列）概念
2. 核心概念：
   • Producer-Consumer 模式：將工作分為生產與消費兩階段，以緩衝區解耦並平衡進度
   • Bounded Blocking Queue：具容量上限的佇列，滿時 Enqueue 阻塞、空時 Dequeue 阻塞
   • 執行緒同步與訊號：用 ManualResetEvent 控制等待/喚醒（WaitOne/Set/Reset）
   • Shutdown 機制：停止新進件、讓消費端消化完庫存後自然結束
   • IO-bound 與 CPU-bound 解耦：跨資源型態（網路IO vs CPU壓縮）並行提升效率
3. 技術依賴：
   • BlockQueue 依賴基礎 Queue 作為內部儲存
   • 阻塞行為依賴 ManualResetEvent（兩個事件：_enqueue_wait、_dequeue_wait）
   • 執行緒安全依賴 lock 於共享資源（_inner_queue）存取
   • Shutdown 流程依賴狀態旗標 _IsShutdown 與事件喚醒，並沿用原生 Queue 的例外行為
4. 應用場景：
   • 批次下載與即時壓縮：下載（IO-bound）與壓縮（CPU-bound）並行
   • 任務管線（Pipeline）：多階段處理串接，以多個 BlockQueue 串聯
   • 背景工作處理：記錄寫入、匯出、影像處理、ETL
   • 控制資源使用：限制記憶體/暫存空間（以容量上限平衡生產與消費速率）

學習路徑建議

1. 入門者路徑：
   • 了解 Producer-Consumer 問題與有界緩衝區概念
   • 練習 C# Thread 基本用法、lock 及 Queue 基本操作
   • 以單一 Producer/Consumer + 小容量 BlockQueue 實作並觀察阻塞行為
2. 進階者路徑：
   • 熟悉 ManualResetEvent 的 Set/Reset/WaitOne 行為與記憶體可見性
   • 嘗試不同執行緒數量組合（P>C、P<C）以觀察吞吐與等待
   • 重構為 Circular Queue、比較與現有 ConcurrentQueue/BlockingCollection
3. 實戰路徑：
   • 將 BlockQueue 應用到實際任務（如下載壓縮或影像轉檔）
   • 加入 Shutdown、錯誤處理、取消（CancellationToken）與日誌
   • 延伸為多階段 Pipeline（多個 BlockQueue 串接），或加入 Priority Queue

關鍵要點清單

• Producer-Consumer 模式：用緩衝區解耦前後兩階段並行處理（優先級: 高）
• Bounded Queue（容量上限）：限制庫存，避免無界成長與資源壓力（優先級: 高）
• 阻塞式 Enqueue/Dequeue：滿/空時暫停等待而非丟例外（優先級: 高）
• ManualResetEvent 訊號控制：Set/Reset/WaitOne 協調生產與消費（優先級: 高）
• lock 保護臨界區：對 _inner_queue 的安全存取（優先級: 高）
• Shutdown 機制：停止新入列、讓消費端清庫存並自然結束（優先級: 高）
• IO-bound vs CPU-bound 解耦：跨資源類型並行以提升吞吐（優先級: 中）
• 執行緒數量調參：P>C 時排隊、P<C 時即時消費，驗證平衡（優先級: 中）
• 例外行為策略：Shutdown 後禁止 Enqueue；空佇列 Dequeue 依原生行為丟例外（優先級: 中）
• 內部事件對偶：_enqueue_wait 與 _dequeue_wait 的互斥喚醒邏輯（優先級: 中）
• Circular Queue 優化：固定容量結構更適合有界緩衝（優先級: 低）
• Priority Queue 變體：高優先權可插隊，適合差異化服務（優先級: 低）
• ThreadPool vs Thread：大批獨立小任務可用 ThreadPool，否則自管執行緒（優先級: 中）
• Pipeline 擴展：多階段處理以多個 BlockQueue 串接（優先級: 中）
• 觀測與測試：透過日誌/Console 觀察阻塞與吞吐以調整容量與並行度（優先級: 中）