从‘Hello World’到高并发：用C# Concurrent集合（ConcurrentBag, ConcurrentDictionary）搞定多线程数据共享

从‘Hello World’到高并发：用C# Concurrent集合搞定多线程数据共享

第一次在C#中遇到多线程数据共享问题时，我盯着屏幕上那个诡异的NullReferenceException发了半小时呆——明明单线程测试一切正常，为什么一开多线程就崩溃？这个经历让我意识到，从单线程思维到并发编程的跨越，远不止是Task.Run()那么简单。本文将带你从零构建一个真实的后台任务处理系统，逐步揭示传统集合在多线程环境中的陷阱，以及如何用System.Collections.Concurrent命名空间下的神器（特别是ConcurrentBag和ConcurrentDictionary）优雅解决这些问题。

1. 为什么我们需要线程安全集合？

假设你正在开发一个电商促销系统，需要实时统计来自全国各地的秒杀请求。单线程版本简单直接：

var requestCounts = new Dictionary<string, int>(); void ProcessRequest(string region) { if (!requestCounts.ContainsKey(region)) requestCounts[region] = 0; requestCounts[region]++; }

这段代码在单线程环境下完美运行，但当我们尝试用并行处理提升性能时：

Parallel.For(0, 1000, _ => ProcessRequest("华东")); Console.WriteLine(requestCounts["华东"]); // 输出可能小于1000！

三个致命问题会突然出现：

1. 丢失更新：多个线程同时执行requestCounts[region]++时，增量操作可能被覆盖
2. 键值竞争：ContainsKey检查和后续赋值不是原子操作
3. 状态损坏：内部数据结构可能因并发修改而崩溃

传统解决方案是使用lock：

private static readonly object _lock = new object(); void ProcessRequestSafe(string region) { lock (_lock) { if (!requestCounts.ContainsKey(region)) requestCounts[region] = 0; requestCounts[region]++; } }

但锁机制存在明显缺陷：

此时就该ConcurrentDictionary登场了——它内部采用细粒度锁和无锁算法的混合策略，既保证线程安全又维持高性能。

2. ConcurrentDictionary实战：电商实时统计系统

让我们重构之前的统计系统：

var concurrentCounts = new ConcurrentDictionary<string, int>(); void ProcessRequestConcurrent(string region) { concurrentCounts.AddOrUpdate(region, 1, (_, count) => count + 1); }

这个简单改动带来了质的飞跃：

性能对比测试（处理100万次请求）：

更妙的是，ConcurrentDictionary提供了一系列原子操作：

// 原子化获取或添加 var value = concurrentCounts.GetOrAdd("华北", key => GetInitialCountFromDB(key)); // 尝试更新避免竞争 concurrentCounts.TryUpdate("华南", 100, 50); // 只有当当前值为50时才更新为100 // 批量操作仍保持线程安全 foreach (var kvp in concurrentCounts) { Console.WriteLine($"{kvp.Key}: {kvp.Value}"); }

实际项目中的经验法则：

• 当读写比例大于10:1时，优先考虑ConcurrentDictionary
• 对于配置数据等读多写少场景，可以结合Lazy<T>实现延迟初始化
• 使用ToArray()获取快照避免枚举时的并发修改异常

3. ConcurrentBag：无序但高效的生产者-消费者模式

在开发日志收集系统时，我遇到过这样的需求：多个线程产生日志条目，单个消费者线程批量处理。最初尝试用ConcurrentQueue：

var logQueue = new ConcurrentQueue<LogEntry>(); // 生产者 void ProduceLog(LogEntry entry) => logQueue.Enqueue(entry); // 消费者 void ProcessLogs() { while (logQueue.TryDequeue(out var entry)) { SaveToDatabase(entry); } }

但当生产者远多于消费者时，ConcurrentQueue的严格FIFO特性反而成为瓶颈。这时ConcurrentBag的线程本地存储设计就大放异彩：

var logBag = new ConcurrentBag<LogEntry>(); // 生产者 void ProduceLog(LogEntry entry) => logBag.Add(entry); // 消费者 void ProcessLogs() { while (logBag.TryTake(out var entry)) { SaveToDatabase(entry); } }

两种集合的微观差异：

一个真实案例：在视频转码服务中，使用ConcurrentBag管理空闲的编码器实例，使线程总能快速获取最近使用过的编码器，相比队列方案减少了30%的初始化开销。

4. 高级模式：组合使用并发集合

成熟的分布式系统往往需要组合多种并发集合。比如在开发实时交易风控系统时，我们设计了这样的架构：

// 风险事件管道 var eventQueue = new ConcurrentQueue<RiskEvent>(); // 用户行为计数器 var userCounters = new ConcurrentDictionary<long, AtomicCounter>(); // 规则检查结果缓存 var ruleResults = new ConcurrentDictionary<string, Lazy<RiskRuleResult>>(); // 生产者线程 void OnTransaction(Transaction tx) { eventQueue.Enqueue(new RiskEvent(tx)); userCounters.GetOrAdd(tx.UserId, _ => new AtomicCounter()).Increment(); } // 消费者线程 void ProcessEvents() { while (eventQueue.TryDequeue(out var evt)) { var result = ruleResults.GetOrAdd(evt.RuleKey, key => new Lazy<RiskRuleResult>(() => EvaluateRule(key))); // 使用result.Value... } }

性能优化技巧：

1. 对于计数器场景，可以封装Interlocked操作为AtomicCounter类
2. 使用Lazy<T>避免重复计算的同时保证线程安全初始化
3. 定期用Clear()清理过期数据，但要注意此时需要暂停所有读写

public class AtomicCounter { private int _count; public void Increment() => Interlocked.Increment(ref _count); public int Value => Volatile.Read(ref _count); }

5. 陷阱与最佳实践

即使使用并发集合，仍有需要注意的深坑：

内存泄漏陷阱：

var userSessions = new ConcurrentDictionary<long, UserSession>(); // 错误示范：永远不会移除旧会话 userSessions.TryAdd(userId, new UserSession()); // 正确做法：定期清理或使用WeakReference

枚举的线程安全幻觉：

foreach (var item in concurrentCollection) { // 虽然不会抛出异常，但此时的item可能已经被其他线程修改 // 对一致性要求高的场景应该使用ToArray()快照 }

容量规划建议：

在最近的一个物联网项目中，我们将设备状态存储在ConcurrentDictionary中，初始容量设置为设备数量的120%，并发级别设置为Environment.ProcessorCount * 2，相比默认设置提升了40%的吞吐量。