C# BlockingCollection

JusterZhu

发布于 2025-01-23 20:40:26

3700

代码可运行

文章被收录于专栏：JusterZhuJusterZhu

运行总次数：0

代码可运行

在多线程编程中，数据共享和线程同步是两个关键问题。C# 提供了一些强大的工具来帮助开发人员管理这些挑战，其中之一就是 BlockingCollection<T>。这篇博客将介绍 BlockingCollection 的基本概念、用法以及它在多线程环境中的优势。

什么是 BlockingCollection？

BlockingCollection<T> 是 .NET 中的一个线程安全集合类，通常用于生产者-消费者模式。它是一个高级的集合类，内部使用了 IProducerConsumerCollection<T> 接口，可以让多个线程安全地添加和移除数据。

为什么使用 BlockingCollection？

线程安全：BlockingCollection 内部实现了锁机制，确保在多线程环境中操作集合时不会出现竞争条件。
阻塞操作：提供了阻塞的 Add 和 Take 方法，允许线程在集合为空或已满时等待。
取消支持：支持通过 CancellationToken 来取消等待操作。
限量容量：可以设置集合的最大容量，控制数据生产速度，防止过度使用内存。

关键方法

Add(T item)：向集合中添加元素，如果集合已满，线程会阻塞。
Take()：从集合中移除元素，如果集合为空，线程会阻塞。
CompleteAdding()：通知集合不再接受新元素。
TryAdd(T item, TimeSpan timeout)：尝试在指定时间内添加元素。
TryTake(out T item, TimeSpan timeout)：尝试在指定时间内移除元素。

基本用法

下面是一个简单的例子，演示如何使用 BlockingCollection<T> 实现生产者-消费者模式。

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

public class Program
{
    public static void Main()
    {
        // 创建一个容量为5的 BlockingCollection
        BlockingCollection<int> blockingCollection = new BlockingCollection<int>(5);

        // 启动生产者任务
        Task producer = Task.Run(() =>
        {
            for (int i = 0; i < 10; i++)
            {
                blockingCollection.Add(i);
                Console.WriteLine($"Produced: {i}");
                Thread.Sleep(100); // 模拟生产时间
            }
            blockingCollection.CompleteAdding();
        });

        // 启动消费者任务
        Task consumer = Task.Run(() =>
        {
            while (!blockingCollection.IsCompleted)
            {
                try
                {
                    int item = blockingCollection.Take();
                    Console.WriteLine($"Consumed: {item}");
                }
                catch (InvalidOperationException)
                {
                    // 当集合完成添加且为空时抛出
                    break;
                }
                Thread.Sleep(150); // 模拟消费时间
            }
        });

        Task.WaitAll(producer, consumer);
    }
}

代码详解

创建集合：使用 BlockingCollection<int>(5) 创建一个最大容量为 5 的集合。
生产者任务：生产者不断向集合中添加数据。当集合已满时，Add 方法会阻塞，直到有空余位置。
消费者任务：消费者从集合中移除数据。当集合为空时，Take 方法会阻塞，直到有新数据可用。
CompleteAdding：调用 CompleteAdding 方法表示不再有新数据添加，消费者可以结束。

注意事项

异常处理：在消费者中处理 InvalidOperationException，用于捕获集合完成添加后的访问。
性能：BlockingCollection 是基于锁的集合，适用于需要线程安全但不要求极高性能的场景。

如果有新的数据持续进入，我们可以使用循环或其他控制机制来处理不断到达的数据，而不使用 CompleteAdding，直到系统关闭为止。

改进的代码示例

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

class Program
{
    static void Main()
    {
        // 创建一个无上限的 BlockingCollection
        BlockingCollection<int> dataQueue = new BlockingCollection<int>();

        // 启动消费者线程
        Task consumerTask = Task.Run(() =>
        {
            while (true)
            {
                // 检查是否有数据可消费
                if (dataQueue.TryTake(out int data, Timeout.Infinite))
                {
                    // 模拟数据处理
                    Console.WriteLine($"Processing data: {data}");
                    Thread.Sleep(100); // 模拟处理延迟
                }
            }
        });

        // 启动多个生产者线程
        Task[] producerTasks = new Task[3];
        for (int i = 0; i < producerTasks.Length; i++)
        {
            int producerId = i;
            producerTasks[i] = Task.Run(() =>
            {
                for (int j = 0; j < 15; j++)
                {
                    int data = producerId * 100 + j;
                    dataQueue.Add(data);
                    Console.WriteLine($"Sensor {producerId} produced: {data}");
                    Thread.Sleep(20); // 模拟数据生成延迟
                }
            });
        }

        // 等待所有生产者完成
        Task.WaitAll(producerTasks);

        // 这里不调用 CompleteAdding，因为我们模拟的是持续数据流

        // 让消费者运行一段时间
        Thread.Sleep(5000); // 运行5秒后停止

        // 停止消费者线程
        consumerTask.Wait();
    }
}