C# 高级教程 - 08 并发集合与同步

多线程环境下，普通的 List<T>、Dictionary<TKey, TValue> 不是线程安全的，在并发读写时会导致数据损坏或异常。.NET 提供了 System.Collections.Concurrent 命名空间，包含一系列专为高并发场景设计的集合类型。本文将深入比较各类并发集合的内部实现、适用场景以及备选的无锁方案。

并发集合概览

类型	非并发等价	特点
`ConcurrentDictionary<TKey, TValue>`	`Dictionary`	细粒度锁 + 无锁读取
`ConcurrentQueue<T>`	`Queue`	无锁（CAS）实现
`ConcurrentStack<T>`	`Stack`	无锁（CAS）实现
`ConcurrentBag<T>`	无	线程本地存储优化
`BlockingCollection<T>`	无	生产者-消费者边界阻塞
`IProducerConsumerCollection<T>`	接口	所有并发集合的公共接口

ConcurrentDictionary

基本操作

var dict = new ConcurrentDictionary<string, int>();

// 线程安全的添加/更新
dict.TryAdd("key1", 1); // 成功返回 true，键已存在返回 false

dict["key2"] = 2; // 存在则更新，不存在则添加

// 原子操作
dict.AddOrUpdate("key1",
    addValueFactory: _ => 100,
    updateValueFactory: (_, old) => old + 1);

dict.GetOrAdd("key3", _ => ExpensiveCreateValue());

// 原子删除
dict.TryRemove("key2", out int removedValue);

内部实现

ConcurrentDictionary 使用分段锁（.NET Core 3.0+ 改用无锁读取 + 细粒度锁写入）实现高并发：

底层基于多个独立 bucket 组（table），每个 table 拥有独立的锁
读操作大部分无锁（使用 Volatile.Read）
写操作仅锁住对应的 table 段

// 特性场景：高性能缓存
public class AsyncCache<TKey, TValue> where TKey : notnull
{
    private readonly ConcurrentDictionary<TKey, Lazy<Task<TValue>>> _cache = new();

    public Task<TValue> GetOrAddAsync(TKey key, Func<TKey, Task<TValue>> factory)
    {
        return _cache.GetOrAdd(key, _ => new Lazy<Task<TValue>>(
            () => factory(key), LazyThreadSafetyMode.ExecutionAndPublication
        )).Value;
    }
}

使用 Lazy<Task<T>> 确保了工厂方法只执行一次，不会出现缓存击穿问题。

ConcurrentQueue

内部使用无锁算法（CAS + 链表），是一种高性能的 FIFO 队列。

var queue = new ConcurrentQueue<int>();

// 入队
queue.Enqueue(1);
queue.Enqueue(2);
queue.Enqueue(3);

// 出队（非阻塞）
if (queue.TryDequeue(out int result))
{
    Console.WriteLine(result); // 1
}

// 查看队首但不移除
if (queue.TryPeek(out int peek))
{
    Console.WriteLine(peek); // 2
}

// 长度（O(1)）
int count = queue.Count;

应用场景：任务调度、日志缓冲区、消息分发的 FIFO 管道。

ConcurrentStack

LIFO 结构，同样使用无锁 CAS 实现。

var stack = new ConcurrentStack<int>();

stack.Push(1);
stack.PushRange([2, 3, 4]); // 批量压入

if (stack.TryPop(out int item))
{
    Console.WriteLine(item); // 4
}

if (stack.TryPeek(out int top))
{
    Console.WriteLine(top); // 3
}

// 批量弹出（减少 CAS 竞争）
int[] items = new int[3];
int popped = stack.TryPopRange(items);

ConcurrentBag

无序集合，针对每个线程独立操作的场景进行了优化。

var bag = new ConcurrentBag<int>();

bag.Add(1);
bag.Add(2);
bag.Add(3);

if (bag.TryTake(out int taken))
{
    Console.WriteLine(taken); // 不确定（线程本地优化导致无序）
}

内部机制：每个线程维护一个本地队列，TryTake 优先从当前线程的本地队列取出元素，仅当本地队列为空时才窃取其他线程的数据。这使得在单生产者/单消费者模式下几乎零竞争。

BlockingCollection

在 IProducerConsumerCollection<T> 之上增加边界阻塞语义，是最经典的生产者-消费者模型实现。

var collection = new BlockingCollection<int>(boundedCapacity: 5);

// 生产者
Task producer = Task.Run(() =>
{
    for (int i = 0; i < 10; i++)
    {
        collection.Add(i); // 队列满时阻塞
        Console.WriteLine($"生产: {i}");
        Thread.Sleep(100);
    }
    collection.CompleteAdding(); // 通知消费者不再生产
});

// 消费者
Task consumer = Task.Run(() =>
{
    foreach (var item in collection.GetConsumingEnumerable())
    {
        Console.WriteLine($"消费: {item}");
        Thread.Sleep(200);
    }
});

await Task.WhenAll(producer, consumer);

多个生产者-消费者

var orders = new BlockingCollection<string>(boundedCapacity: 20);

// 3 个生产者
var producers = Enumerable.Range(0, 3).Select(i => Task.Run(() =>
{
    for (int j = 0; j < 10; j++)
    {
        orders.Add($"订单 {i}-{j}");
        Thread.Sleep(Random.Shared.Next(50, 150));
    }
}));

// 2 个消费者
var consumers = Enumerable.Range(0, 2).Select(i => Task.Run(() =>
{
    foreach (var order in orders.GetConsumingEnumerable())
    {
        Console.WriteLine($"消费者 {i} 处理: {order}");
        Thread.Sleep(Random.Shared.Next(100, 300));
    }
}));

await Task.WhenAll(producers);
orders.CompleteAdding();
await Task.WhenAll(consumers);

Channel（.NET Core 3.0+）

System.Threading.Channels.Channel<T> 是现代 .NET 中更推荐的生产者-消费者方案，兼具高性能和 async-first 设计。

var channel = System.Threading.Channels.Channel.CreateBounded<int>(10);

// 生产者
Task producer = Task.Run(async () =>
{
    for (int i = 0; i < 10; i++)
    {
        await channel.Writer.WriteAsync(i);
        await Task.Delay(100);
    }
    channel.Writer.Complete();
});

// 消费者
Task consumer = Task.Run(async () =>
{
    await foreach (var item in channel.Reader.ReadAllAsync())
    {
        Console.WriteLine($"收到: {item}");
        await Task.Delay(200);
    }
});

Channel vs BlockingCollection

特性	Channel	BlockingCollection
异步 API	✅ `WriteAsync/ReadAsync`	❌ 仅同步 `Add/Take`
性能	更高（ValueTask + 无锁路径）	一般（锁 + Monitor.Wait）
取消支持	原生 CancellationToken	需自行处理
单一/多消费者	支持	支持

不可变集合（Immutable Collections）

对于读多写极少的场景，.NET 提供了 System.Collections.Immutable：

using System.Collections.Immutable;

var builder = ImmutableArray<int>.Empty.ToBuilder();
builder.Add(1);
builder.Add(2);
builder.Add(3);
var array = builder.ToImmutable(); // 不可变快照

var dict = ImmutableDictionary<string, int>.Empty
    .Add("one", 1)
    .Add("two", 2);

// 修改产生新实例（结构共享）
var newDict = dict.SetItem("one", 10);
Console.WriteLine(dict["one"]);    // 1（原对象不变）
Console.WriteLine(newDict["one"]); // 10

// 线程安全 —— 无需锁即可并发读取
ImmutableList<int> shared = ImmutableList<int>.Empty;

Parallel.For(0, 100, i =>
{
    Interlocked.CompareExchange(ref shared,
        shared.Add(i), shared); // 原子替换
});

性能对比指南

场景	推荐集合	原因
高频读、写较少	`ConcurrentDictionary`	无锁读
FIFO 顺序重要	`Channel<T>` 或 `ConcurrentQueue`	高吞吐
线程本地数据为主	`ConcurrentBag`	低竞争
需要阻塞生产者	`BlockingCollection` 或 `Channel`	流量控制
读极多、写极少	`ImmutableDictionary` + 引用替换	零锁
跨进程同步	`MemoryMappedFile` + `Mutex`	OS 级别

常见陷阱

// 陷阱 1: 在并发字典的值上使用非线程安全操作
var dict = new ConcurrentDictionary<string, List<int>>();
dict.GetOrAdd("key", _ => new List<int>()).Add(1); // List<int> 不是线程安全的！
// 正确做法：使用不可变集合或锁住值

// 陷阱 2: 枚举时修改
foreach (var kvp in dict) // ConcurrentDictionary 的快照枚举是安全的
{
    // 但如果你枚举的同时更新值，得到的可能是旧值或新值
}

// 陷阱 3: Count 不代表当前精确值
int count = dict.Count; // 其他线程可能正在添加/删除

总结：并发集合让多线程数据共享变得安全且高效。选择正确的并发集合可以大幅减少锁竞争、避免死锁，并提升应用吞吐量。理解每种集合的内部布局和无锁算法原理，是做出正确选型的关键。在 .NET Core 的新项目中，Channel<T> 正逐渐成为生产者-消费者场景的首选方案。