進階的模式與 Concurrent 集合的特性

1. 「生產者‑消費者」 + Concurrent

我們之前在討論 ConcurrentQueue 時已經提過生產者‑消費者模式。現在來更細緻地看，有多個生產者與消費者時，以及如何正確地發出結束訊號。

ConcurrentQueue（和其他 Concurrent 集合）的最大優勢是它自己處理線程安全。你不用把 Enqueue 或 TryDequeue 用 lock 包起來——多個執行緒可以安全地透過共同的佇列互動。

範例：多個生產者與多個消費者

多個 worker 執行緒會產生任務，另外幾個執行緒處理這些任務。

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

ConcurrentQueue<string> taskQueue = new ConcurrentQueue<string>();
CancellationTokenSource cts = new CancellationTokenSource(); // 用來取消消費者的工作

// 生產者方法
void Producer(string name, int count)
{
    for (int i = 0; i < count; i++)
    {
        string task = $"Zadacha_{name}_{i}";
        taskQueue.Enqueue(task);
        Console.WriteLine($"[P:{name}] Dobavil: {task}");
        Thread.Sleep(10); 
    }
}

// 消費者方法
void Consumer(string name)
{
    while (!cts.Token.IsCancellationRequested || taskQueue.Count > 0)
    {
        if (taskQueue.TryDequeue(out string task))
        {
            Console.WriteLine($"[C:{name}] Obrabotal: {task}");
            Thread.Sleep(20); 
        }
        else
        {
            Thread.Sleep(50); // 等待，如果佇列是空的
        }
    }
    Console.WriteLine($"[C:{name}] Zavershil rabotu.");
}

// 在 Main 中啟動範例:
Task.Run(() => Producer("A", 10)); // 生產者 A
Task.Run(() => Producer("B", 10)); // 生產者 B
Task.Run(() => Consumer("1"));    // 消費者 1
Task.Run(() => Consumer("2"));    // 消費者 2

Thread.Sleep(1000); // 給足夠時間運作
cts.Cancel();       // 向消費者發送結束工作訊號
Thread.Sleep(500); // 給消費者時間處理剩餘項並結束

在這個例子裡，多個生產者與消費者同時操作同一個 ConcurrentQueue，不會發生資料競爭：Enqueue 和 TryDequeue 都是原子操作。

結束訊號的重要性（CancellationTokenSource）

我們用 CancellationTokenSource（簡稱 cts）來告訴消費者應該結束工作。這對 Producer‑Consumer 模式很關鍵：

• 生產者已經完成工作。 當不再有新增項目時，消費者不應該無限期地等待空佇列。
• 應用程式要關閉。 需要能正確地停止消費者。

CancellationTokenSource 與 CancellationToken 提供了標準化的機制：消費者週期性檢查 IsCancellationRequested，必要時可呼叫 ThrowIfCancellationRequested()。

2. BlockingCollection<T>

雖然 ConcurrentQueue<T> 很適合做生產者‑消費者，但它需要你自己在空佇列時等待，以及手動發出結束信號。若要更方便的實作，.NET 提供了 BlockingCollection<T>——它不是獨立的集合，而是包在任何 IProducerConsumerCollection<T>（例如 ConcurrentQueue）之上。

BlockingCollection 的優點：

• 阻塞操作。 Add()/Take() 會在集合滿或空時阻塞執行緒。不需要手動檢查 IsEmpty。
• 限制大小。 可以設置容量 (Capacity)。當達到限制時 Add() 會被阻塞——對於記憶體控制很有用。
• 方便的完成處理。 CompleteAdding() 用來表示不會再新增，GetConsumingEnumerable() 允許消費者在整個生產完成前持續處理。

範例：用 BlockingCollection<T> 的 Producer‑Consumer

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

// 預設 BlockingCollection 使用 ConcurrentQueue
BlockingCollection<int> numbers = new BlockingCollection<int>(capacity: 10); // 容量 10

void ProducerBC(int count)
{
    for (int i = 0; i < count; i++)
    {
        numbers.Add(i); // 若佇列滿會阻塞
        Console.WriteLine($"[P] Dobavil: {i}");
        Thread.Sleep(50);
    }
    numbers.CompleteAdding(); // 通知不再新增
    Console.WriteLine("[P] Proizvoditel zavershil dobavlenie.");
}

void ConsumerBC()
{
    // GetConsumingEnumerable 會阻塞，只要還有元素或尚未呼叫 CompleteAdding
    foreach (var item in numbers.GetConsumingEnumerable())
    {
        Console.WriteLine($"[C] Obrabotal: {item}");
        Thread.Sleep(100);
    }
    Console.WriteLine("[C] Potrebitel zavershil rabotu.");
}

// 在 Main 中啟動範例:
Task producerTask = Task.Run(() => ProducerBC(15)); // 15 個元素，容量 10
Task consumerTask = Task.Run(ConsumerBC);
Task.WaitAll(producerTask, consumerTask); // 等待完成

注意：因為有 GetConsumingEnumerable()，消費者端的程式碼變得更乾淨。如果你需要阻塞操作或容量限制，BlockingCollection 是個好工具。

3. 其他 Concurrent 集合的方法與屬性

IsEmpty, Count

• IsEmpty (bool)：集合是否為空。
• Count (int)：當前元素數量。

範例：使用 IsEmpty 和 Count

using System.Collections.Concurrent;

ConcurrentQueue<string> q = new ConcurrentQueue<string>();
Console.WriteLine($"Ochered pustaya? {q.IsEmpty}"); // True

q.Enqueue("A");
q.Enqueue("B");
Console.WriteLine($"Elementov v ocheredi: {q.Count}"); // 2
Console.WriteLine($"Ochered pustaya? {q.IsEmpty}"); // False

q.TryDequeue(out var itemA);
Console.WriteLine($"Elementov v ocheredi: {q.Count}"); // 1

轉為陣列（ToArray()）

所有 Concurrent 集合都有 ToArray() 方法，會回傳元素的一個瞬間快照。

範例：使用 ToArray()

using System.Collections.Concurrent;

ConcurrentStack<int> s = new ConcurrentStack<int>();
s.Push(10);
s.Push(20);
s.Push(30);

int[] items = s.ToArray(); // 會建立一個新陣列: [30, 20, 10] (對於堆疊 LIFO)
Console.WriteLine($"Elementy v masive: {string.Join(", ", items)}");

// 集合本身不會改變
Console.WriteLine($"Elementov v steke posle ToArray: {s.Count}"); // 3

清空集合

在 .NET 6+ 許多 Concurrent 集合新增了 Clear() 方法來移除所有元素。

範例：清空集合

using System.Collections.Concurrent;

ConcurrentBag<string> bag = new ConcurrentBag<string>();
bag.Add("Alpha");
bag.Add("Beta");
Console.WriteLine($"Elementov v Bag: {bag.Count}"); // 2

bag.Clear(); // 清空集合
Console.WriteLine($"Elementov v Bag posle ochistki: {bag.Count}"); // 0
Console.WriteLine($"Bag pust? {bag.IsEmpty}"); // True

4. Concurrent 集合行為的注意事項

要記得「瞬間快照」的概念。像是 Count 這類屬性和 ToArray() 的結果反映的是某一時刻的狀態。在並行改變的情況下，這個值可能在取得後立刻過時。

範例：Count 與 ToArray() —「瞬間快照」

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

ConcurrentQueue<int> snapshotQueue = new ConcurrentQueue<int>();

void AddItemsContinuously()
{
    for (int i = 0; i < 1000; i++)
    {
        snapshotQueue.Enqueue(i);
        Thread.Sleep(1); 
    }
}

// 在 Main 啟動範例:
Task.Run(AddItemsContinuously); // 一個不斷加入項目的執行緒

Thread.Sleep(100); // 給一點時間加入
Console.WriteLine($"Tekushchiy Count: {snapshotQueue.Count}"); // 可能是 50、80、120...
Thread.Sleep(100);
Console.WriteLine($"Tekushchiy Count snova: {snapshotQueue.Count}"); // 會是另一個值
int[] currentItems = snapshotQueue.ToArray();
Console.WriteLine($"Kolichestvo elementov v ToArray(): {currentItems.Length}"); // 可能和前面的 Count 不同

不要把 Count 當成在活躍修改期間的絕對保證。

5. 在 Concurrent 集合上迭代的細節

單一操作（像是 Add、TryTake、TryPop、GetOrAdd 等）是線程安全的。但如果你用 foreach 去迭代一個可能被其他執行緒修改的集合，並不保證你會看到所有元素或只看到某一時刻的元素——可能會遺漏或出現意外行為。

範例：在修改期間迭代

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

ConcurrentQueue<int> iterQueue = new ConcurrentQueue<int>();

// 加入初始元素
for (int i = 0; i < 10; i++) iterQueue.Enqueue(i);

// 修改者執行緒
void Modifier()
{
    for (int i = 10; i < 20; i++)
    {
        iterQueue.Enqueue(i); // 新增元素
        Thread.Sleep(50);
    }
}

// 迭代者執行緒
void Iterator()
{
    Console.WriteLine("Nachinaem iteraciyu...");
    int count = 0;
    foreach (var item in iterQueue) // 迭代
    {
        Console.Write($"{item} ");
        count++;
        Thread.Sleep(30); // 模擬工作，讓修改者有機會改變集合
    }
    Console.WriteLine($"\nIteraciya zavershena. Prochteno {count} elementov.");
    Console.WriteLine($"Tekuschee kolichestvo v ocheredi: {iterQueue.Count}");
}

// 在 Main 中啟動範例:
Task.Run(Modifier);
Task.Run(Iterator);
Thread.Sleep(1500); // 給足夠時間運作

規則：如果你需要一個固定的元素集（例如用於報表），先用 ToArray() 取得快照，然後再對那個陣列做迭代：

// 如果集合可能改變，這是正確的迭代方式
int[] snapshot = iterQueue.ToArray();
foreach (var item in snapshot)
{
    // 現在你是在一個不可變的陣列快照上迭代
}

這就是我們對 Concurrent 集合進階模式與特性的介紹：看了多角色的 Producer‑Consumer 與 BlockingCollection，還有使用 Count、ToArray() 與迭代時應注意的重點。