介绍 Concurrent 集合

1. 问题的来龙去脉

在单线程应用里，像 List<T>、Dictionary<T> 这样的集合行为是可预测的。但一旦多个线程同时访问同一个集合，就会出现熟悉的问题：数据竞争（race conditions）。

如果多个线程在没有正确同步的情况下同时读写同一个集合，你可能会遇到：

• 数据不正确：某个线程删除了元素，而另一个线程正尝试更新它。
• 数据丢失：一个线程添加了元素，另一个线程在不知道之前添加的情况下覆盖了它。
• 异常：集合可能变成不合法的状态，你会看到 InvalidOperationException（比如 "Collection was modified; enumeration operation may not execute."）甚至 NullReferenceException。

示例 1：在 List<T> 中的数据竞争（简单的自增）

两个线程同时对列表的同一个元素做自增。

using System.Collections.Generic;
using System.Threading.Tasks; // 用于 Task.Run

class RaceConditionExample
{
    static List<int> numbers = new List<int> { 0 }; // 一个元素的列表

    static void Main(string[] args)
    {
        Console.WriteLine("初始值: " + numbers[0]); // 0

        // 启动两个线程，每个线程都对 numbers[0] 做自增
        Task task1 = Task.Run(() => IncrementNumbers(500_000));
        Task task2 = Task.Run(() => IncrementNumbers(500_000));

        Task.WaitAll(task1, task2); // 等待两个线程完成

        Console.WriteLine("结束值: " + numbers[0]); // 期望 1_000_000，但...
        // 结果几乎总是小于 1_000_000！
    }

    static void IncrementNumbers(int count)
    {
        for (int i = 0; i < count; i++)
        {
            // 这个操作 "numbers[0]++" 实际上由 3 个步骤组成：
            // 1. 读取 numbers[0]
            // 2. 值加 1
            // 3. 把新值写回 numbers[0]
            numbers[0]++; 
        }
    }
}

为什么这是数据竞争？如果线程 A 读到了 numbers[0]（值为 0），然后在 A 写回 1 之前，线程 B 也读到了 0，那么两个线程都会把 0 增到 1 并写回 1。一次自增就丢失了。操作 numbers[0]++ 不是原子的。

示例 2：在遍历 Dictionary 时触发 InvalidOperationException

一个线程在遍历字典，另一个线程修改它。

using System.Collections.Generic;
using System.Threading; // 用于 Thread.Sleep

class DictionaryRaceExample
{
    static Dictionary<int, string> users = new Dictionary<int, string>();

    static void Main(string[] args)
    {
        // 初始化字典
        for (int i = 0; i < 5; i++) users.Add(i, $"User {i}");

        // 读线程
        Thread readerThread = new Thread(() =>
        {
            try
            {
                foreach (var user in users) // 遍历字典
                {
                    Console.WriteLine($"读取者: {user.Key} - {user.Value}");
                    Thread.Sleep(10); // 模拟工作
                }
            }
            catch (InvalidOperationException ex)
            {
                Console.WriteLine($"读取者: 错误！ {ex.Message}");
            }
        });

        // 写线程
        Thread writerThread = new Thread(() =>
        {
            Thread.Sleep(5); // 给读取者一点时间开始
            for (int i = 5; i < 10; i++)
            {
                users.Add(i, $"New User {i}"); // 添加元素
                Console.WriteLine($"写入者: 添加了 User {i}");
                Thread.Sleep(15);
            }
        });

        readerThread.Start();
        writerThread.Start();

        readerThread.Join(); // 等待线程结束
        writerThread.Join();
        Console.WriteLine("示例结束。");
    }
}

为什么会出错？Dictionary<TKey, TValue>（和 List<T> 一样）并不是设计成在没有同步的情况下被不同线程同时读写的。当写线程改变内部结构时，读线程继续在已经被修改的数据上做 foreach，这会导致 InvalidOperationException。

2. 为什么简单的 lock 不总是最优？

把所有东西都用 lock 包起来看起来很简单，但有缺点：

// 糟糕的示例：锁的粒度太大
// （仅作演示，不建议这样做！）
static object _lock = new object();
static List<int> _sharedList = new List<int>();

void AddItem(int item)
{
    lock (_lock)
    {
        _sharedList.Add(item);
    }
}

int GetItemCount()
{
    lock (_lock)
    {
        return _sharedList.Count;
    }
}

• 性能（瓶颈）： lock 会阻塞对整个集合的访问。在 100 个线程的情况下，99 个会等着 1 个，即使很多操作本来不会冲突。
• 复杂性：你得在每个访问集合的地方记得加 lock。一处遗漏——数据竞争又回来了。
• 死锁：在不同对象上使用多个 lock 很容易导致 deadlock。
• 迭代器：即便你用 foreach，如果另一个线程修改了集合，也没用。

所以在 .NET 里引入了专门的线程安全集合。

原子操作

线程安全集合 —— 能保证在多个线程并发访问时也能正确工作，用户不需要在外部自己加锁。关键是 原子操作：要么整个操作完成，要么不做，其他线程不会看到“中间态”。

• 添加、删除、读取看起来像是按顺序一个接一个执行的。
• 内部会用到底层技术：互锁操作（Interlocked）、Compare-And-Swap (CAS)、更细粒度的锁——而不是对整个集合的全局锁。

3. System.Collections.Concurrent 概览

命名空间 System.Collections.Concurrent 提供了一组为了并发而从头设计的集合。它们的理念是 最大化并行，最小化锁。

• 性能：随着核数增加能更好地扩展。
• 简单：不需要在每次操作外面手动加 lock。
• 更少的错误：避免了很多手动同步带来的问题。
• 为竞争场景优化：在并发添加/删除时效率更高。

4. 主要类

ConcurrentQueue<T>（线程安全队列）

原则：FIFO — 先进先出。场景：producer–consumer、日志、任务队列。

using System.Collections.Concurrent;

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

void Producer() => messageQueue.Enqueue("消息 1");

void Consumer()
{
    if (messageQueue.TryDequeue(out string message))
    {
        Console.WriteLine($"已处理: {message}");
    }
    else
    {
        Console.WriteLine("队列为空。");
    }
}

ConcurrentStack<T>（线程安全栈）

原则：LIFO — 后进先出。场景：操作历史、DFS 遍历、对象池。

using System.Collections.Concurrent;

ConcurrentStack<int> historyStack = new ConcurrentStack<int>();

void PushAction(int value) => historyStack.Push(value);

void PopAction()
{
    if (historyStack.TryPop(out int action))
    {
        Console.WriteLine($"已撤销操作: {action}");
    }
    else
    {
        Console.WriteLine("栈为空。");
    }
}

ConcurrentBag<T>（线程安全“袋子”）

无序集合，不保证顺序。针对“线程更可能取到自己放进去的东西”这种场景做了优化。很适合对象池。

using System.Collections.Concurrent;

ConcurrentBag<System.Guid> objectPool = new ConcurrentBag<System.Guid>();

void AddObject() => objectPool.Add(System.Guid.NewGuid());

void TakeObject()
{
    if (objectPool.TryTake(out System.Guid obj))
    {
        Console.WriteLine($"取到对象: {obj}");
    }
    else
    {
        Console.WriteLine("对象池为空。");
    }
}

ConcurrentDictionary<TKey, TValue>（线程安全字典）

支持原子的添加、更新和按键获取值。很适合缓存、会话、计数器。

using System.Collections.Concurrent;

ConcurrentDictionary<string, int> userScores = new ConcurrentDictionary<string, int>();

void UpdateScore(string user, int score)
{
    // 如果不存在就原子添加，存在则原子更新
    userScores.AddOrUpdate(user, score, (key, existingVal) => existingVal + score);
    Console.WriteLine($"分数 {user}: {userScores[user]}");
}

void GetScore(string user)
{
    if (userScores.TryGetValue(user, out int score))
    {
        Console.WriteLine($"当前分数 {user}: {score}");
    }
    else
    {
        Console.WriteLine($"用户 {user} 未找到。");
    }
}

5. 什么时候用这些集合替代普通集合？

• 应用是多线程的：在只有一个线程的时候普通集合更快（没有额外开销）。
• 多个线程共享一个集合：这是选择 System.Collections.Concurrent 的关键信号。
• 需要高性能和可扩展性：这些集合为减小等待而设计。
• 想让代码更简单：不想在每次操作周围写手动的 lock。
• 需要原子操作：添加/删除/获取不会把集合留下不一致的状态。

以下情况不要用 Concurrent 集合：

• 应用严格是单线程的。
• 你需要对多次相关操作做“事务式”处理（可能需要外部同步或其它机制）。
• 你依赖严格的提取顺序，而这种顺序在某些集合（比如 ConcurrentBag<T>）中不能保证。