C#ConcurrentBag使用详解

总目录

---

前言

在多线程编程中，如何安全地共享数据是一个常见的挑战。C# 提供了多种线程安全的集合类型来解决这个问题，其中之一就是 ConcurrentBag<T>。本文将详细介绍 ConcurrentBag<T> 的使用方法、适用场景及其优缺点。

---

一、 基本信息 1. 核心概念 在多线程编程中，线程安全集合是避免数据竞争和保证高效协作的核心工具。ConcurrentBag<T> 是 .NET Framework 4.0+ 引入的线程安全集合，属于 System.Collections.Concurrent 命名空间。专为多线程频繁添加/获取元素的场景设计。它实现了无序的、线程安全的添加和移除操作。与 List<T> 或 Queue<T> 不同，ConcurrentBag<T> 不保证元素的顺序，并且每个线程都可以高效地添加和移除自己的元素。 2. 核心特性 线程安全：无需手动加锁，内部通过细粒度锁或无锁技术（如 ThreadLocal）实现高效并发。无序集合：元素没有固定顺序，不保证元素的插入顺序，TryTake 可能返回任意元素，适合不需要顺序的场景。高性能：针对“同一线程既添加又移除元素”的场景做了优化（例如线程本地存储），减少竞争。 二、基础用法与示例 1. 初始化与添加元素 using System; using System.Collections.Concurrent; class Program { static void Main() { var bag = new ConcurrentBag<int>(); // 添加元素 bag.Add(1); bag.Add(2); bag.Add(3); Console.WriteLine("Elements in the bag:"); foreach (var item in bag) { Console.WriteLine(item); } } } 遍历 ConcurrentBag 中的所有元素可以通过 foreach 循环实现 static void Main() { ConcurrentBag<int> bag = new ConcurrentBag<int>(); // 多线程并发添加 Parallel.For(0, 10, i => { bag.Add(i); Console.WriteLine($"Thread {Environment.CurrentManagedThreadId} 添加了 {i}"); }); } 2. 取出/移除元素（TryTake）

TryTake 方法用于尝试从 ConcurrentBag 中移除一个元素。如果成功，则返回 true 并将元素赋值给输出参数；否则返回 false。

// 尝试取出元素（若为空返回false） bool success = bag.TryTake(out int result); if (success) { Console.WriteLine($"Successfully removed: {result}"); } else { Console.WriteLine("Failed to remove an element."); } 3. 查看但不移除（TryPeek） if (bag.TryPeek(out int peekItem)) { Console.WriteLine($"查看元素（不移除）: {peekItem}"); } 4. Count

获取当前元素数量（高并发下可能不精确，慎用逻辑判断）。

5. bool IsEmpty

判断集合是否为空（同样存在瞬时性，可能不准确）。

三、为什么使用ConcurrentBag？ 1. 非线程安全示例 class Program { static void Main() { // 非线程安全版本（错误示例） var unsafeList = new List<int>(); Parallel.For(0, 1000, i => { unsafeList.Add(i); // 会导致数据丢失或抛出异常 }); Console.WriteLine($"非安全集合数量: {unsafeList.Count}"); // 结果通常小于1000 } }

运行结果：

运行代码时，unsafeList.Count 通常会远小于 1000（例如 900~999），甚至可能抛出异常。结果不确定：由于线程竞争是随机的，每次运行的结果可能不同。 2. 为啥不安全？ 1）问题根源 List<T> 内部结构：List<int> 内部通过数组实现，当调用 Add() 时，会检查数组容量。如果容量不足，会触发扩容操作（创建新数组并复制旧数据）。扩容操作不是原子性的。并发写入冲突：在 Parallel.For 中，多个线程可能同时调用 Add()，导致以下问题： 数据覆盖：多个线程可能同时写入同一个数组位置，导致数据丢失。扩容竞争：线程 A 触发扩容时，线程 B 可能仍在操作旧数组，引发索引越界或数据不一致。计数器错误：List 的 Count 属性通过一个内部计数器维护，多线程同时修改会导致计数器值不准确。 2）具体错误场景示例

场景 1 - 数据丢失： 假设两个线程同时执行 Add(i)，它们的操作可能如下：

Thread 1: 读取当前 Count = 5 → 写入索引5 → 更新 Count = 6 Thread 2: 读取当前 Count = 5 → 写入索引5 → 更新 Count = 6

结果：索引5被覆盖，实际只添加了一个元素，但 Count 错误地更新了两次。

场景 2 - 扩容异常： 若多个线程同时触发扩容，可能导致内部数组被多次重建，最终引发 IndexOutOfRangeException 或数据错乱。

3. 解决方案 1）显式加锁（性能较低） var lockObj = new object(); Parallel.For(0, 1000, i => { lock (lockObj) { unsafeList.Add(i); } }); 2）使用线程安全集合 var safeList = new ConcurrentBag<int>(); // 线程安全集合 Parallel.For(0, 1000, i => { safeList.Add(i); }); 四、典型应用场景 1. 多线程任务处理

ConcurrentBag 在多线程环境中特别有用。以下是一个简单的例子，展示了如何在多个线程中同时向 ConcurrentBag 中添加元素。

using System; using System.Collections.Concurrent; using System.Threading.Tasks; class Program { static void Main() { var bag = new ConcurrentBag<int>(); var tasks = new Task[5]; for (int i = 0; i < tasks.Length; i++) { tasks[i] = Task.Run(() => { for (int j = 0; j < 1000; j++) { bag.Add(j); } }); } Task.WaitAll(tasks); Console.WriteLine($"Total elements in the bag: {bag.Count}"); } } 2. 工作窃取

工作窃取（Work Stealing）是一种并发编程模式，允许线程从其他线程的任务队列中“窃取”任务来执行，从而提高 CPU 的利用率和整体性能。

每个线程优先处理自己的本地队列，空闲时可“窃取”其他线程的任务，减少竞争。

using System; using System.Collections.Concurrent; using System.Threading; using System.Threading.Tasks; class Program { static void Main() { // 创建一个 ConcurrentBag 实例，用于线程安全的任务存储 ConcurrentBag<int> taskBag = new ConcurrentBag<int>(); // 初始化任务：向任务池中添加10个任务 for (int i = 1; i <= 10; i++) { taskBag.Add(i); } // 创建3个消费者线程，模拟工作窃取 Task[] workers = new Task[3]; for (int i = 0; i < workers.Length; i++) { workers[i] = Task.Run(() => { // 每个线程不断尝试从任务池中获取任务 while (taskBag.TryTake(out int task)) { // 模拟任务处理（此处用随机延迟） var random = new Random(); Thread.Sleep(random.Next(50, 200)); Console.WriteLine($"线程 {Task.CurrentId} 处理任务: {task}"); } }); } // 等待所有线程完成任务 Task.WaitAll(workers); Console.WriteLine("所有任务处理完毕！"); } }

代码说明：

ConcurrentBag 初始化 通过 ConcurrentBag<int> 创建一个线程安全的任务池，并预填充10个任务。 工作窃取机制 多个线程（此处为3个）并发执行，通过 TryTake 方法从任务池中获取任务。当某个线程的本地队列没有任务时，会从其他线程的队列中“窃取”任务（由 ConcurrentBag 内部自动实现）。 输出示例 运行结果类似如下（实际顺序可能因线程调度而异）： 线程 4 处理任务: 1 线程 3 处理任务: 3 线程 4 处理任务: 5 线程 3 处理任务: 7 线程 5 处理任务: 2 ... 所有任务处理完毕！ 3. 临时数据收集

并行计算时各线程生成中间结果，最后合并到 ConcurrentBag。

ConcurrentBag<double> results = new(); Parallel.For(0, 1000, i => results.Add(Calculate(i))); using System; using System.Collections.Concurrent; using System.Threading.Tasks; class Program { static void Main() { // 创建一个并发集合 ConcurrentBag<int> results = new ConcurrentBag<int>(); // 并行处理数据 Parallel.For(0, 100, i => { int processedValue = ProcessData(i); results.Add(processedValue); // 多线程安全添加 }); // 输出结果数量 Console.WriteLine($"Collected {results.Count} results."); } static int ProcessData(int input) { // 模拟数据处理 return input * 2; } }

为什么适合 ConcurrentBag？

多个线程同时向集合添加数据。无需关心元素顺序，只需收集所有结果。线程安全且性能优于手动加锁的 List。 五、注意事项与最佳实践 1. 注意事项

无序性：ConcurrentBag 不保证元素的插入顺序，取出元素的顺序不可预测，如果你的应用需要有序的数据访问，请考虑其他线程安全的集合类型，如 ConcurrentQueue<T> 或 ConcurrentStack<T>。

性能权衡：

虽然 ConcurrentBag 提供了良好的并发性能，但在某些情况下（例如频繁的跨线程操作，如一个线程存、另一个线程取），其性能可能不如预期。此时考虑 ConcurrentQueue 或 ConcurrentStack。相比传统的非线程安全集合（如 List 或 Dictionary<TKey, TValue>），ConcurrentBag 在多线程环境下提供了更高的性能和更好的线程安全性。然而，如果顺序对应用至关重要，建议使用其他并发集合（如 ConcurrentQueue 或 ConcurrentDictionary<TKey, TValue>）

空集合处理 TryTake 可能失败，需结合循环或超时机制：

while (!bag.IsEmpty) { if (bag.TryTake(out int item)) Process(item); } 2. 最佳实践 优先使用线程本地操作：让每个线程尽可能处理自己添加的元素。避免频繁跨线程操作：减少不同线程间的数据交互。合理选择集合类型：需要有序访问时改用 ConcurrentQueue/ConcurrentStack。 3. 适用场景 最佳场景：最适合“线程既生产又消费数据”的情况（例如对象池），若纯生产者-消费者分离，可能其他集合更高效。任务分配：在多个线程之间分配任务，任务的顺序不重要高并发环境下的临时存储/结果收集：当需要在多个线程之间共享数据，你的应用程序只需要快速添加和移除元素，但对顺序没有要求时，ConcurrentBag<T> 是一个很好的选择。工作窃取：多个消费者线程可以从同一个 ConcurrentBag 中“窃取”任务。生产者-消费者模式：尤其适合生产者-消费者模式。 六、与其他线程安全集合对比 集合类型顺序性适用场景ConcurrentBag<T>无序线程频繁添加/取出自己的元素ConcurrentQueue<T>FIFO严格的先进先出队列ConcurrentStack<T>LIFO后进先出（类似栈操作）BlockingCollection<T>可配置带容量限制的阻塞式集合

---

结语

回到目录页：C#/.NET 知识汇总 希望以上内容可以帮助到大家，如文中有不对之处，还请批评指正。

---

参考资料： 官方文档参考 Microsoft Learn - ConcurrentBag 第九章：并发集合