C#.NET ConcurrentBag<T> 设计原理与使用场景

graph TD
    A[线程A] --> A_Queue[本地队列A: 1, 3, 5]
    B[线程B] --> B_Queue[本地队列B: 2, 4]
    C[线程C] --> C_Queue[本地队列C: 6]
    
    D[全局队列] --> |工作窃取| A_Queue
    D --> |工作窃取| B_Queue
    D --> |工作窃取| C_Queue

var bag = new ConcurrentBag<string>();

// 插入（极快）
bag.Add("任务A");
bag.Add("任务B");

// 尝试取出（非阻塞）
if (bag.TryTake(out var item))
{
    Console.WriteLine($"取出: {item}");
}

// 尝试偷取（TryPeek 不存在！）
if (bag.TryTake(out var stolen)) { /* 处理 */ }

// 计数（注意：有一定开销）
int count = bag.Count;

// 清空（不常用）
bag.Clear();

// 检查是否为空
bool isEmpty = bag.IsEmpty;

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

class ConcurrentBagBasicDemo
{
    static void Main()
    {
        // 创建线程安全的ConcurrentBag
        var bag = new ConcurrentBag<int>();

        // 1. 多线程添加元素（4个线程，每个添加5个元素）
        Parallel.For(0, 4, threadId =>
        {
            for (int i = 1; i <= 5; i++)
            {
                int value = threadId * 100 + i;
                bag.Add(value);
                Console.WriteLine($"线程{threadId}：添加 {value}");
            }
        });

        Console.WriteLine($"\n集合总元素数：{bag.Count}\n");

        // 2. 多线程消费元素（直到集合为空）
        Parallel.For(0, 2, threadId =>
        {
            while (!bag.IsEmpty)
            {
                if (bag.TryTake(out int value))
                {
                    Console.WriteLine($"线程{threadId}：取出 {value}");
                }
                // 避免空循环占用CPU
                Task.Delay(10).Wait();
            }
        });

        Console.WriteLine($"\n最终集合是否为空：{bag.IsEmpty}");
    }
}

var files = Directory.GetFiles("big_folder", "*.txt");
var bag = new ConcurrentBag<string>(files);

Parallel.ForEach(bag, new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount },
    file =>
    {
        ProcessFile(file);
    });

public class ObjectPool<T>
{
    private readonly ConcurrentBag<T> _objects;
    private readonly Func<T> _objectGenerator;

    public ObjectPool(Func<T> objectGenerator)
    {
        _objects = new ConcurrentBag<T>();
        _objectGenerator = objectGenerator;
    }

    public T Get()
    {
        return _objects.TryTake(out T item) ? item : _objectGenerator();
    }

    public void Return(T item)
    {
        _objects.Add(item);
    }
}

// 使用示例
var pool = new ObjectPool<StringBuilder>(() => new StringBuilder());
var sb = pool.Get();
try
{
    sb.Append("Hello");
    Console.WriteLine(sb.ToString());
}
finally
{
    pool.Return(sb);
}

   // 同一线程添加和取出
   var threadLocalBag = new ConcurrentBag<WorkItem>();
   
   void Process()
   {
       threadLocalBag.Add(CreateWork());
       if (threadLocalBag.TryTake(out var work))
       {
           Execute(work);
       }
   }

   // 生产者消费者分离
   var sharedBag = new ConcurrentBag<Data>();
   
   // 生产者线程
   Task.Run(() => sharedBag.Add(produce()));
   
   // 消费者线程
   Task.Run(() => 
   {
       if (sharedBag.TryTake(out var data))
       {
           consume(data);
       }
   });