佇列與堆疊 Producer-Consumer

1. 介紹

從基礎開始。佇列 是一個基本資料結構，遵循 FIFO (First-In, First-Out) 原則，也就是「先到先出」。想像超市裡的排隊：先到的人先被服務。

在多執行緒程式設計中，模式 生產者-消費者 (Producer-Consumer) 是最常見且強大的設計之一。

• 生產者 (Producers) — 指會建立資料或任務並把它們放到共用佇列的執行緒或應用程式部分。他們「生產」工作。
• 消費者 (Consumers) — 指從佇列取出資料或任務並處理它們的執行緒或應用程式部分。他們「消費」工作。

這個模式有助於控制資料流、解耦系統組件（生產者不需要知道誰或如何處理資料）、提升系統響應性，並幫助在執行緒間平均分配負載。

範例： ConcurrentQueue — 新增與取出

看看如何對 ConcurrentQueue<T> 新增與取出元素。

using System.Collections.Concurrent;

ConcurrentQueue<string> tasks = new ConcurrentQueue<string>();

// 新增元素（生產者）
tasks.Enqueue("下載檔案");
tasks.Enqueue("處理影像");
Console.WriteLine($"佇列中的任務數: {tasks.Count}"); // 輸出: 佇列中的任務數: 2

// 取出元素（消費者）
if (tasks.TryDequeue(out string task1))
{
    Console.WriteLine($"已完成任務: {task1}"); // 輸出: 已完成任務: 下載檔案
}

if (tasks.TryDequeue(out string task2))
{
    Console.WriteLine($"已完成任務: {task2}"); // 輸出: 已完成任務: 處理影像
}

if (!tasks.TryDequeue(out string emptyTask))
{
    Console.WriteLine("佇列為空，沒有新任務。"); // 輸出: 佇列為空，沒有新任務。
}

基礎操作： Enqueue(), TryDequeue()

Enqueue(T item): 用來將元素新增到佇列尾端。這個操作是執行緒安全的。你可以同時在 10 個不同執行緒調用 Enqueue，所有元素都會正確加入。

TryDequeue(out T item): 用來嘗試從佇列頭部取出元素。這是消費者的關鍵方法。若成功取出元素（值放到輸出參數 item），它回傳 true；若佇列為空，回傳 false。重要的是，TryDequeue 不會在佇列為空時阻塞執行緒。

2. TryDequeue() 的重要性與操作原子性

方法 TryDequeue() 不只是方便；它對正確的執行緒安全工作是至關重要的。它是原子性的：檢查佇列是否為空與實際取出元素是作為一個不可分割的操作執行。

如果我們只有單獨的方法 IsEmpty（檢查佇列是否為空）和 Dequeue（取出元素），那在兩者之間，其他執行緒可能會把佇列清空。結果你的 Dequeue 會拋出例外或返回錯誤資料。TryDequeue 完全避免了這種情況。

範例：多執行緒的 Producer-Consumer

這裡我們啟動兩個生產者執行緒與一個消費者執行緒。

using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

ConcurrentQueue<int> dataQueue = new ConcurrentQueue<int>();
bool producersDone = false; // 用於通知消費者的旗標

void Producer(int start, int count)
{
    for (int i = 0; i < count; i++)
    {
        dataQueue.Enqueue(start + i);
        Console.WriteLine($"[P] 新增: {start + i}");
        Thread.Sleep(10); 
    }
}

void Consumer()
{
    while (!producersDone || dataQueue.Count > 0) // 當還有資料或生產者在工作時持續
    {
        if (dataQueue.TryDequeue(out int item))
        {
            Console.WriteLine($"[C] 處理: {item}");
        }
        else
        {
            Thread.Sleep(50); // 佇列為空時等一會
        }
    }
    Console.WriteLine("[C] 完成工作。");
}

// 在 Main 中啟動範例:
// Task.Run(() => Producer(1, 5));
// Task.Run(() => Producer(100, 5)); // 第二個生產者
// Task.Run(() => Consumer());
// Thread.Sleep(600); // 給執行緒一些時間工作
// producersDone = true; // 通知生產者已完成
// Thread.Sleep(200); // 給消費者時間取出剩餘項目

注意，在這個簡單範例中使用旗標 producersDone 與 Thread.Sleep 來模擬結束。在真實應用中，更可靠的結束同步通常會使用 CancellationTokenSource 或 BlockingCollection<T>。

ConcurrentQueue<T> 非常適合下列情境：

• 處理順序很重要（FIFO）。
• 多個執行緒新增元素，和/或多個執行緒取出元素。
• 需要高效能且不想手動管理鎖。

3. 用於 producer-consumer 的堆疊 (LIFO)

堆疊 是另一個基本資料結構，遵循 LIFO (Last-In, First-Out) 原則，也就是「後進先出」。想像一疊盤子：你總是拿最上面的盤子，新增的盤子放在最上面。

ConcurrentStack<T> 與 ConcurrentQueue<T> 一樣是執行緒安全的，也可以用在「生產者-消費者」模式，但處理順序相反。

範例： ConcurrentStack — 新增與取出

using System.Collections.Concurrent;

ConcurrentStack<string> commandStack = new ConcurrentStack<string>();

// 新增命令（生產者）
commandStack.Push("選取文字");
commandStack.Push("更改字型");
commandStack.Push("儲存文件");
Console.WriteLine($"堆疊中的命令數: {commandStack.Count}"); // 輸出: 堆疊中的命令數: 3

// 取出命令（消費者）
if (commandStack.TryPop(out string cmd1))
{
    Console.WriteLine($"已還原命令: {cmd1}"); // 輸出: 已還原命令: 儲存文件
}

if (commandStack.TryPop(out string cmd2))
{
    Console.WriteLine($"已還原命令: {cmd2}"); // 輸出: 已還原命令: 更改字型
}

if (!commandStack.TryPop(out string emptyCmd))
{
    Console.WriteLine("命令堆疊為空。"); // 輸出: 命令堆疊為空。
}

4. 基本操作： Push(), TryPop()

Push(T item): 用來將元素新增到堆疊頂端。這個操作是執行緒安全的。

TryPop(out T item): 用來嘗試從堆疊頂端取出元素。若成功取出，回傳 true；若堆疊為空，回傳 false。跟 TryDequeue 一樣，這是原子操作，可避免資料競爭。

範例：使用 ConcurrentStack 實作物件池

堆疊很適合實作物件池：拿出 — 使用 — 放回。

using System.Collections.Concurrent;

class Connection { /* 簡單的替代 */ public Guid Id { get; } = Guid.NewGuid(); }

ConcurrentStack<Connection> connectionPool = new ConcurrentStack<Connection>();

// 用初始連線填充池
for (int i = 0; i < 3; i++)
{
    connectionPool.Push(new Connection());
}
Console.WriteLine($"池中的連線數: {connectionPool.Count}"); // 輸出: 池中的連線數: 3

void UseConnection()
{
    if (connectionPool.TryPop(out Connection conn))
    {
        Console.WriteLine($"[Pool] 使用連線: {conn.Id}");
        // 模擬使用連線
        Thread.Sleep(50); 
        connectionPool.Push(conn); // 放回池中
        Console.WriteLine($"[Pool] 已歸還連線: {conn.Id}. 池中數量: {connectionPool.Count}");
    }
    else
    {
        Console.WriteLine("[Pool] 連線池為空。建立新連線。");
        // 通常在此處建立新的連線，若池為空
        connectionPool.Push(new Connection()); 
    }
}

// 在 Main 中啟動範例:
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Thread.Sleep(500);

在這個範例中，數個執行緒可以安全地從共用池中取得並歸還連線。

5. 使用範例與與 ConcurrentQueue 的比較

ConcurrentStack<T> 適用於：

• 需要 LIFO 的處理順序（例如「還原」操作的歷史記錄）。
• 需要快速存取最近新增的元素（通常在 CPU 快取中仍然「熱」）。
• 實作以堆疊為基礎的演算法（深度優先搜尋 (DFS)、語法解析等）。

比較與選擇合適的集合

ConcurrentQueue

ConcurrentStack

在你的 producer-consumer 場景中，選擇 ConcurrentQueue 或 ConcurrentStack 完全取決於你需要的元素處理順序。這兩種集合都提供開箱即用的高效能與執行緒安全，讓你不必為手動同步煩惱，幫助構建可擴充的多執行緒系統。

2

任務

C# SELF, 等級 58, 課堂 1

上鎖

使用佇列實現 Producer-Consumer

使用佇列實現 Producer-Consumer

開啟