ConcurrentBag 和 ConcurrentDictionary

1. 介绍

ConcurrentBag<T> 是一个线程安全的、无序集合。它最大的特点和优势体现在单词“Bag”（袋子）上，这意味着在取出元素时不保证顺序。也就是说，你取出的元素可能不是按添加顺序所期望的那个。相比之下，ConcurrentBag 在某些场景下有独特的优化，使它非常快。

ConcurrentBag 的特点

无序： 与队列（FIFO）和栈（LIFO）不同，ConcurrentBag 不保证 TryTake() 会按任何特定顺序返回元素。这个是它的关键区别。

针对本地访问的优化 (Thread-Local Storage)： ConcurrentBag 存在的主要原因就是在这样一种场景下具有高性能：添加元素的线程很可能就是随后取出该元素的线程。

示例：ConcurrentBag — 添加和取出

using System.Collections.Concurrent;

ConcurrentBag<string> itemBag = new ConcurrentBag<string>();

// 添加元素
itemBag.Add("条目 A");
itemBag.Add("条目 B");
itemBag.Add("条目 C");

Console.WriteLine($"袋中元素数量: {itemBag.Count}"); // 输出: 袋中元素数量: 3

// 取出元素（顺序不保证！）
if (itemBag.TryTake(out string item1))
{
    Console.WriteLine($"已取出: {item1}"); // 可能是 "条目 C", "条目 B" 或 "条目 A"
}

if (itemBag.TryTake(out string item2))
{
    Console.WriteLine($"已取出: {item2}");
}

Console.WriteLine($"剩余元素数量: {itemBag.Count}"); // 输出: 剩余元素数量: 1

你可以多次运行这段代码，会注意到取出元素的顺序可能会变化。

方法 Add(), TryTake()

Add(T item): 用于向 ConcurrentBag 添加元素。该操作是线程安全的。

TryTake(out T item): 尝试从 ConcurrentBag 取出一个元素。如果成功取出则返回 true，如果袋子为空则返回 false。注意，TryTake 不会阻塞线程。

2. 使用场景

ConcurrentBag 不是用来替代 ConcurrentQueue 或 ConcurrentStack。它在一些特定情况下表现出色：

对象/资源池： 当你有一个可重用对象池，并且希望通常返回对象的线程更可能再次取到它时。这样可以减少对共享资源的争用。

在 TPL 中的动态任务分配： 类似 Parallel.ForEach 和 Parallel.For 的内部实现使用本地袋子和“work-stealing”机制来高效分配工作。

使用 ConcurrentBag 的任务池并优化本地性

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

ConcurrentBag<string> taskPool = new ConcurrentBag<string>();

// 用初始任务填充池
for (int i = 0; i < 10; i++)
{
    taskPool.Add($"任务 {i}");
}

void Worker()
{
    // 每个线程都尝试取任务
    while (taskPool.TryTake(out string task))
    {
        Console.WriteLine($"线程 {Thread.CurrentThread.ManagedThreadId}: 处理 {task}");
        Thread.Sleep(50); // 模拟工作
    }
    Console.WriteLine($"线程 {Thread.CurrentThread.ManagedThreadId}: 完成工作。");
}

// 启动几个工作线程
// Task.Run(Worker);
// Task.Run(Worker);
// Task.Run(Worker);
// Thread.Sleep(1000); // 给点时间执行

在这个例子里，ConcurrentBag 允许线程高效地取任务，借助内部结构最小化锁竞争。

内部机制

ConcurrentBag 通过使用线程本地存储 (Thread-Local Storage) 提升性能。当线程添加元素时，元素会放到该线程本地的结构里。调用 TryTake() 时会先读取本地结构；如果本地为空，则会从其他线程或全局池执行“work-stealing”。这减少了竞争，对于访问本地性重要且不关心顺序的场景来说，ConcurrentBag 是很好的选择。

3. 线程安全的字典

ConcurrentDictionary<TKey, TValue> 是最常用的线程安全集合之一：用于在多个线程间安全地添加、读取、更新和删除键值对的高性能字典。

普通的 Dictionary<TKey, TValue> 完全不是线程安全的。任何写操作（添加/修改/删除），甚至在写入期间的读取，都可能导致异常 (InvalidOperationException) 或数据损坏。

示例：普通 Dictionary 的问题（复习）

using System.Collections.Generic;
using System.Threading.Tasks;

Dictionary<int, int> concurrentDictProblem = new Dictionary<int, int>();

void AddToDict(int start, int count)
{
    for (int i = 0; i < count; i++)
    {
        // 尝试同时添加/修改
        // 会导致异常或不正确的行为
        concurrentDictProblem[start + i] = start + i;
    }
}

// 在 Main 中运行示例:
try
{
    Task t1 = Task.Run(() => AddToDict(0, 10000));
    Task t2 = Task.Run(() => AddToDict(5000, 10000)); // 键有重叠
    Task.WaitAll(t1, t2);
    Console.WriteLine($"（有问题）字典中的元素数量: {concurrentDictProblem.Count}");
}
catch (Exception ex)
{
    Console.WriteLine($"普通字典错误: {ex.Message}");
}

这段代码几乎肯定会由于线程安全问题抛出异常或死锁。

4. 主要操作

ConcurrentDictionary 提供了一些专门的原子操作来实现“检查 + 操作”的语义。

TryAdd(TKey key, TValue value): 原子地添加键值对。如果键被添加则返回 true，如果键已存在则返回 false。

ConcurrentDictionary<string, int> scores = new ConcurrentDictionary<string, int>();
if (scores.TryAdd("Alice", 100))
    Console.WriteLine("Alice 已添加."); // 输出: Alice 已添加.
if (!scores.TryAdd("Alice", 150))
    Console.WriteLine("Alice 已存在."); // 输出: Alice 已存在.

TryGetValue(TKey key, out TValue value): 原子地按键获取值。

if (scores.TryGetValue("Alice", out int aliceScore))
    Console.WriteLine($"Alice 的分数: {aliceScore}"); // 输出: Alice 的分数: 100

TryUpdate(TKey key, TValue newValue, TValue comparisonValue): 原子地更新值，只有当当前值等于 comparisonValue 时才会更新。用于防止竞态条件。

// 当前 Alice 的值 = 100
if (scores.TryUpdate("Alice", 120, 100)) // 将 100 更新为 120
    Console.WriteLine("Alice 的分数已更新为 120."); // 输出: Alice 的分数已更新为 120.
if (!scores.TryUpdate("Alice", 130, 100)) // 不会更新，因为当前值是 120，而不是 100
    Console.WriteLine("Alice 的分数未更新（数据已过时）。"); // 输出: ...

TryRemove(TKey key, out TValue value): 原子地按键移除元素。

if (scores.TryRemove("Alice", out int removedScore))
    Console.WriteLine($"Alice 已删除，分数为: {removedScore}"); // 输出: Alice 已删除，分数为: 120

5. 进阶原子操作

下面两个方法是 ConcurrentDictionary 的主力，能覆盖很多场景。

GetOrAdd(TKey key, TValue valueFactory(TKey key)): 原子地返回已有的值，或者通过工厂方法创建并添加新的值。非常适合缓存和唯一实体的场景。

// 假设我们缓存一些重量级对象
ConcurrentDictionary<int, HeavyObject> objectCache = new ConcurrentDictionary<int, HeavyObject>();

HeavyObject GetOrCreateHeavyObject(int id)
{
    // 如果已存在 — 返回它，否则通过工厂创建并添加
    return objectCache.GetOrAdd(id, (key) =>
    {
        Console.WriteLine($"为 ID 创建新的 HeavyObject: {key}");
        return new HeavyObject(key); // 模拟创建昂贵对象
    });
}

// 在 Main 中:
HeavyObject obj1 = GetOrCreateHeavyObject(1); // 会创建新的
HeavyObject obj2 = GetOrCreateHeavyObject(2); // 会创建新的
HeavyObject obj3 = GetOrCreateHeavyObject(1); // 会返回已存在的 obj1

AddOrUpdate(TKey key, TValue addValue, Func<TKey, TValue, TValue> updateValueFactory): 原子地添加值（如果键不存在），或通过工厂函数更新已有值。

• addValue: 如果键不存在时用来添加的值。
• updateValueFactory: 基于键和当前值计算新值的函数。

// 统计页面访问量
ConcurrentDictionary<string, int> pageViews = new ConcurrentDictionary<string, int>();

void IncrementPageView(string page)
{
    pageViews.AddOrUpdate(page, 1, // 如果是新页面，添加 1
                          (key, existingVal) => existingVal + 1); // 否则加 1
    Console.WriteLine($"页面 '{page}' 被访问了 {pageViews[page]} 次。");
}

// 在 Main:
IncrementPageView("Home");   // Home: 1
IncrementPageView("About");  // About: 1
IncrementPageView("Home");   // Home: 2
IncrementPageView("Home");   // Home: 3
IncrementPageView("Contact"); // Contact: 1

6. 缓存或状态管理的使用示例

数据缓存： ConcurrentDictionary 是内存缓存的优秀选择：使用 GetOrAdd 可以避免重复创建昂贵对象。

管理用户会话： 在多个请求中安全地存储和更新会话数据。

统计计数： 使用 AddOrUpdate 很方便地递增事件计数、页面浏览量、投票数等。

注册表/Service Locator： 存储已注册的服务或插件，以便从不同线程访问。

ConcurrentDictionary<TKey, TValue> 是一个高度优化的集合，通过一组原子操作大大简化了多线程环境下对字典的操作，无需手动同步。