потокобезопасных паттернов

1. "無鎖" (Lock-Free / Wait-Free) 模式

你已經知道 Concurrent 集合是 потокобезопасны 的。但它們是怎麼做到不用用 lock 來包裹你的代碼就能安全運行的呢？答案藏在一些高級算法裡，通常被稱為 lock-free（無鎖）或甚至 wait-free（無等待）算法。

簡要解釋 lock-free 和 wait-free 算法的概念

Lock-Free（無鎖）:

核心思想：保證至少一個線程能夠每次都向前推進一步，即使其他線程遇到延遲或中斷。

與 lock 的差別：使用 lock 時，如果一個線程持有鎖，其他想要獲取同一個鎖的線程就得等待。而在 lock-free 算法中，線程不會等待，它們會嘗試執行操作，如果遇到衝突，就重試（retry）。

例子：想像你在排隊付錢。如果收銀員忙，你就站著等。這是鎖的行為。lock-free 就像你看到收銀員忙，然後再試一次，而不是一直站在原地不動。

Wait-Free（無等待）:

核心思想：比 lock-free 更強的保證。每個線程都能在有限次自己的操作內，確保向前推進一步，無論其他線程在做什麼。也就是說，沒有人會無限等待。

差別：在 lock-free 中，因為衝突，線程可能會不斷重試。而在 wait-free 中，這種情況不會發生。

實踐：實現 wait-free 較為困難，這類算法較少見。大多數 Concurrent 集合會用 lock-free 或混合策略。

2. Concurrent 集合是如何實現 потокобезопасности 的

lock-free 算法的核心是 原子操作，這些操作由處理器原生支持。你已經知道原子性是什麼。在這個層面上，原子操作是處理器能夠作為一個不可分割的整體來執行的指令。在 .NET 中，我們用靜態類 System.Threading.Interlocked 來操作它們。

Interlocked 操作：

這些是底層、非常快速的原子操作，適用於基本類型（int、long）。

範例：Interlocked.Increment(ref value)（原子性地增加值），Interlocked.CompareExchange(ref location, value, comparand)（原子性比較並交換：只有當 location 的值等於 comparand 時，才將其設為 value）。

CAS（Compare-And-Swap） - 比較並交換：

CompareExchange 就是 CAS 操作的實現。它是許多 lock-free 算法的基石。

流程：

1. 讀取變數的當前值。
2. 根據這個值做一些計算。
3. 嘗試用 只有在變數仍然保持第一步讀取的值時，將其設為新值。如果變數已被其他線程改變，CAS 會失敗，然後重試（回到第一步）。

範例：簡單的 lock-free 計數器，使用 Interlocked

比較普通的自增操作與 Interlocked.Increment：

using System.Threading;
using System.Threading.Tasks;

class CounterExample
{
    static int regularCounter = 0;
    static int interlockedCounter = 0;

    static void IncrementRegular(int iterations)
    {
        for (int i = 0; i < iterations; i++)
        {
            regularCounter++; // 非 потокобезопасно!
        }
    }

    static void IncrementInterlocked(int iterations)
    {
        for (int i = 0; i < iterations; i++)
        {
            Interlocked.Increment(ref interlockedCounter); // 原子操作!
        }
    }
}

// 在 Main 裡：
Task t1 = Task.Run(() => IncrementRegular(500_000));
Task t2 = Task.Run(() => IncrementRegular(500_000));
Task.WaitAll(t1, t2);
Console.WriteLine($"普通計數器: {regularCounter}"); // 可能少於 1_000_000

regularCounter = 0; // 重置
t1 = Task.Run(() => IncrementInterlocked(500_000));
t2 = Task.Run(() => IncrementInterlocked(500_000));
Task.WaitAll(t1, t2);
Console.WriteLine($"Interlocked 計數器: {interlockedCounter}"); // 一定是 1_000_000

方法 Interlocked.Increment 保證每次遞增操作都完整執行，不會因為多線程同時存取而遺失數據。

3. 為什麼這對擴展性和性能很重要

使用 lock-free 算法（通過 Interlocked 和 CAS）可以：

• 降低開銷：傳統的鎖（lock）可能很昂貴，因為涉及上下文切換、核心等待等操作。lock-free 算法能最小化這些成本。
• 避免死鎖（Deadlocks）：因為線程不會等待鎖，不會形成等待循環。
• 提升擴展性：在多核 CPU 上，lock-free 結構讓線程能更獨立地工作，避免成為瓶頸。核數越多，優勢越明顯。
• 反應更快：線程不會長時間阻塞，系統整體反應速度可能更高。

簡單看一下 ConcurrentQueue 的內部結構

為了理解它的運作方式，我們來看一個簡化版的 ConcurrentQueue<T>。它是用鏈結串列（linked list）實作的，每個元素都包含一個指向下一個元素的引用。

加入元素（Enqueue）的流程：

1. 線程創建一個新節點，並存放元素。
2. 用 Interlocked.CompareExchange 原子性地將「尾巴」指針指向新節點。如果另一個線程已經先行加入，CAS 不會成功，當前線程會重試。

取出元素（TryDequeue）的流程：

1. 線程讀取「頭」指針。
2. 用 CAS 原子性地將「頭」指針移到下一個元素，只有在「頭」未被其他線程改變時才成功。
3. 如果失敗（因為其他線程已經取走了元素），就重試。

這是非常簡化的說明，實際實作會更複雜，還要處理 ABA 問題和垃圾回收等。但核心思想是所有指針和資料的操作都用原子指令來完成，避免了繁重的鎖。

4. Concurrent 集合的性能

與傳統鎖定集合的性能比較

Concurrent 集合通常比用手動 lock 保護的普通集合性能更佳，尤其在高競爭（多線程同時存取）場景下。

低競爭： 當存取頻率低或線程數少時，lock-free 算法的開銷可能比簡單的 lock 更大，差異很小甚至略慢。但在大多數情況下，差距不大。

高競爭： 在多線程「爭奪」同一資源時，傳統鎖會導致等待，形成瓶頸。而 Concurrent 集合能讓線程更獨立地工作，減少等待時間，性能提升明顯。

性能比較範例（概念性，不用實測）

using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Diagnostics; // 用於 Stopwatch
using System.Threading.Tasks;

class PerformanceTest
{
    static List<int> regularList = new List<int>();
    static ConcurrentQueue<int> concurrentQueue = new ConcurrentQueue<int>();
    static object lockObject = new object();

    const int Iterations = 1_000_000;
    const int NumTasks = 4; // 並行任務數

    public static void RunTests()
    {
        Console.WriteLine("性能測試（加入元素）:");

        // 使用普通 List + lock
        regularList.Clear();
        Stopwatch sw = Stopwatch.StartNew();
        Parallel.For(0, NumTasks, (i) =>
        {
            for (int j = 0; j < Iterations / NumTasks; j++)
            {
                lock (lockObject)
                {
                    regularList.Add(j);
                }
            }
        });
        sw.Stop();
        Console.WriteLine($"List + lock: {sw.ElapsedMilliseconds} 毫秒. Count: {regularList.Count}");

        // 使用 ConcurrentQueue
        concurrentQueue.Clear();
        sw = Stopwatch.StartNew();
        Parallel.For(0, NumTasks, (i) =>
        {
            for (int j = 0; j < Iterations / NumTasks; j++)
            {
                concurrentQueue.Enqueue(j);
            }
        });
        sw.Stop();
        Console.WriteLine($"ConcurrentQueue: {sw.ElapsedMilliseconds} 毫秒. Count: {concurrentQueue.Count}");
        // 預期在 NumTasks > 1 時，ConcurrentQueue 會明顯快很多
    }
}

重要結論：如果你的代碼中有用 lock 保護集合，建議轉用 Concurrent 類似的方案。

6. 有用的細節

競爭（contention）對性能的影響

Contention（競爭） 指多個線程同時試圖存取同一個共享資源（比如集合）。競爭越激烈，等待的線程越多，性能就越差。

Concurrent 集合設計時會盡量降低 contention。例如，ConcurrentBag 使用線程本地存儲，讓大部分操作不會互相干擾。ConcurrentDictionary 則用細粒度鎖或 lock-free 算法，讓不同線程可以同時操作不同的「桶」（buckets）。

提升性能的關鍵是減少 contention。 如果設計允許，盡量讓線程操作不同的資料片段或不同的集合。

選擇合適的集合，根據場景

選擇集合不僅是「是否 потокобезопасно」，更是「哪個集合在性能和邏輯上最適合我的場景」。

7. 優化建議

即使用 Concurrent 集合，也有一些「陷阱」可能影響性能，值得注意：

避免頻繁使用 ToArray() 在熱點路徑（hot path）

ToArray() 會產生整個集合的新副本。這是昂貴的操作，尤其是大集合，因為涉及記憶體分配和元素複製。

建議：只在確實需要快照時才用 ToArray()，並盡量少用。不要用它來檢查元素數量（用 Count，但要注意它是瞬時快照）。

小心遍歷 Concurrent 集合

如前所述，Concurrent 集合的迭代器在修改時不一定是安全的。如果在遍歷期間，其他線程持續修改集合，可能會漏掉元素或得到不一致的結果。

建議：如果需要穩定的快照，先用 ToArray() 產生快照，再遍歷這個陣列。

  // 不好：可能漏掉元素或看到變化
  foreach (var item in myConcurrentQueue) { /* ... */ }

  // 好：用快照遍歷
  var snapshot = myConcurrentQueue.ToArray();
  foreach (var item in snapshot) { /* ... */ }

減少「流量」（traffic）

盡量將資料或任務分組，減少單次操作（Add/Take）的次數。操作越少， contention 越低。

例如，不要每個訊息都單獨加入，改成批次加入一大包訊息。

注意 contention

如果性能出現瓶頸，調查是哪個部分的 contention 最嚴重。或許可以改變設計，讓線程處理不同資料片段或不同集合，降低競爭。