C#并发集合Concurrent类详解与BlockingCollection使用示例

一、概述：线程安全集合的重要性

在 .NET 开发中，当多个线程并发访问同一个集合时，直接使用 System.Collections 或 System.Collections.Generic 中的类型（如 List<T>）会导致线程安全隐患。为此，.NET 提供了 System.Collections.Concurrent 命名空间，包含一系列线程安全的集合类。

该命名空间的核心基础是 IProducerConsumerCollection<T> 接口，它定义了生产者-消费者模式的关键操作：

• TryAdd(T item)：尝试向集合添加项。返回布尔值指示操作是否成功，若集合禁止添加（如已标记完成），则失败。
• TryTake(out T item)：尝试从集合中移除并获取项。返回布尔值指示操作结果，成功时将元素赋值给 out 参数。

二、主要并发集合类解析

第1步：理解 ConcurrentQueue - FIFO 队列

目的：实现先进先出（FIFO）的线程安全队列。

内部原理：基于无锁算法，使用内部合并到链表中的固定大小数组，减少锁竞争。

核心方法：

• Enqueue(T item)：入队。
• TryDequeue(out T result)：出队并移除元素。
• TryPeek(out T result)：查看队首元素但不移除。

注意：TryAdd 和 TryTake 实际上分别调用 Enqueue 和 TryDequeue。

第2步：理解 ConcurrentStack - LIFO 堆栈

目的：实现后进先出（LIFO）的线程安全堆栈。

核心方法：

• Push(T item) / PushRange：压栈。
• TryPop(out T result)：弹栈并移除元素。
• TryPopRange：批量弹栈。

注意：底层同样基于链表结构，保证高并发下的性能。

第3步：理解 ConcurrentBag - 无序包

目的：存储无需排序或优先级的元素集合。

特点：不保证取出元素的顺序，内部采用线程映射机制以减少锁开销。

核心方法：Add()、TryTake()、TryPeek()。由于无序性，其性能在某些高频吞吐场景下优于 Queue 和 Stack。

第4步：理解 ConcurrentDictionary - 线程安全字典

目的：提供线程安全的键值对集合。

特殊性：因为基于键值查找而非纯生产者-消费者模式，所以它未实现 IProducerConsumerCollection 接口。

核心原子方法：

• TryAdd(key, value)：若键不存在则添加。
• TryGetValue(key, out value)：安全获取值。
• TryRemove(key, out value)：安全移除。
• TryUpdate(key, newValue, oldValue)：若当前值等于预期值，则更新为新值，否则失败。这是实现 CAS（Compare-And-Swap）操作的关键。

第5步：掌握 BlockingCollection - 阻塞式集合

目的：为任何实现了 IProducerConsumerCollection<T> 的集合提供阻塞和界限控制功能。默认底层使用 ConcurrentQueue<T>。

工作机制：

• 阻塞等待：当生产者试图添加到已满的集合，或消费者试图从空集合取数据时，线程会默认阻塞等待，直到条件满足。
• 取消支持：可以通过传递 CancellationToken 来中断阻塞调用。
• 超时控制：TryAdd 和 TryTake 支持指定超时时间，避免无限期等待。
• 完成信号：调用 CompleteAdding() 通知生产者不再添加新元素，消费者随后会在处理完剩余元素后检测到完成状态。

三、实战：使用 BlockingCollection 实现生产者-消费者模型

以下代码演示了如何使用 BlockingCollection 在两个任务（Task）之间进行线程间通信：一个任务作为生产者生成数据，另一个任务作为消费者处理数据。

internal static BlockingCollection<int> _TestBCollection;

class ThreadWork1  // 生产者
{
    public void Run()
    {
        Console.WriteLine("ThreadWork1 run {");
        for (int i = 0; i < 100; i++)
        {
            Console.WriteLine($"ThreadWork1 producer: {i}");
            _TestBCollection.Add(i); // 如果集合有上限且满，此处会阻塞
        }
        _TestBCollection.CompleteAdding(); // 标记生产结束
        Console.WriteLine("ThreadWork1 run }");
    }
}

class ThreadWork2  // 消费者
{
    public void Run()
    {
        int i = 0;
        int nCnt = 0;
        Console.WriteLine("ThreadWork2 run {");
        // IsCompleted 在所有元素被消费且 Addings 完成后变为 true
        while (!_TestBCollection.IsCompleted)
        {
            bool success = _TestBCollection.TryTake(out i);
            if (success)
            {
                Console.WriteLine($"ThreadWork2 consumer: {i * i} ===== {i}");
                nCnt++;
            }
        }
        Console.WriteLine($"Total items consumed: {nCnt}");
        Console.WriteLine("ThreadWork2 run }");
    }
}

class Program
{
    static void Main(string[] args)
    {
        // 初始化，可设置上限，例如 new BlockingCollection<int>(10)
        _TestBCollection = new BlockingCollection<int>(); 

        Task t1 = Task.Run(() => new ThreadWork1().Run());
        Task t2 = Task.Run(() => new ThreadWork2().Run());

        Console.WriteLine("Main wait t1 t2 end {");
        Task.WaitAll(t1, t2); // 等待所有任务完成
        Console.WriteLine("Main wait t1 t2 end }");
    }
}

代码关键点解析：

1. 数据流同步：生产者通过 Add 加入整数，消费者通过 TryTake 取出。若消费者执行快于生产者，TryTake 会短暂返回 false 或阻塞（取决于具体 API 变体，这里使用的是非阻塞轮询配合 IsCompleted 检查，实际生产中更常用直接阻塞的 Take 方法或带超时的 TryTake）。
2. 优雅退出：生产者必须调用 CompleteAdding()，这样消费者才能通过 IsCompleted 属性知道何时停止循环。
3. Task 封装：使用 Task.Run 替代传统的 Thread，更符合现代 .NET 异步编程最佳实践。

四、总结

在处理多线程并发问题时，选择合适的集合至关重要：ConcurrentQueue/Stack/Bag 适用于简单的无阻塞存取；ConcurrentDictionary 适用于键值对的并发读写；而 BlockingCollection 则是构建复杂生产者-消费者管道、解耦线程逻辑的首选工具。正确使用这些类可以显著简化线程同步代码，提高系统吞吐量与稳定性。

以上就是C#并发集合Concurrent类及BlockingCollection用法的详细内容，更多关于C#多线程编程的资料请关注本站其它相关文章！