簡(jiǎn)介

ConcurrentBag<T> 是 System.Collections.Concurrent 命名空間下的線(xiàn)程安全的無(wú)序集合，專(zhuān)為 “多線(xiàn)程同時(shí)添加 / 移除元素” 設(shè)計(jì)，核心特點(diǎn)是基于線(xiàn)程局部存儲(chǔ)（TLS）優(yōu)化，在 “同一線(xiàn)程頻繁添加和移除元素” 的場(chǎng)景下性能最優(yōu)，是 .NET 中處理無(wú)序線(xiàn)程安全集合的核心工具。

核心定位與價(jià)值

在多線(xiàn)程場(chǎng)景中，普通的 List<T> 非線(xiàn)程安全（多線(xiàn)程操作會(huì)拋出異?；驍?shù)據(jù)損壞），而 lock 包裹的 List<T> 存在鎖競(jìng)爭(zhēng)問(wèn)題（性能低）。ConcurrentBag<T> 的核心價(jià)值：

• 無(wú)鎖核心路徑：通過(guò)線(xiàn)程局部存儲(chǔ)（TLS）讓每個(gè)線(xiàn)程優(yōu)先操作自己的私有數(shù)據(jù)段，減少跨線(xiàn)程鎖競(jìng)爭(zhēng)；
• 無(wú)序存儲(chǔ)：不保證元素的順序（插入順序≠遍歷順序），犧牲順序換取性能；
• 線(xiàn)程安全：所有操作（Add/TryTake 等）均線(xiàn)程安全，無(wú)需手動(dòng)加鎖；
• 適配特定場(chǎng)景：尤其適合 “生產(chǎn)者和消費(fèi)者為同一線(xiàn)程” 的場(chǎng)景（如線(xiàn)程池線(xiàn)程自產(chǎn)自銷(xiāo)）。

核心特點(diǎn)

內(nèi)部實(shí)現(xiàn)原理

ConcurrentBag 的高性能來(lái)源于線(xiàn)程本地存儲(chǔ) + 工作竊取的設(shè)計(jì)：

• 每個(gè)線(xiàn)程擁有一個(gè)私有小袋（bag）（通常是鏈表或數(shù)組）
• 線(xiàn)程 Add/Take 時(shí)優(yōu)先操作自己的私有袋（幾乎無(wú)鎖）
• 當(dāng)自己袋子為空時(shí)，會(huì)去 “偷” 其他線(xiàn)程的袋子（ work-stealing）

這種設(shè)計(jì)導(dǎo)致：

• 同一個(gè)線(xiàn)程插入的元素，很可能被同一個(gè)線(xiàn)程先取出（局部性好）
• 但跨線(xiàn)程看，完全無(wú)序，而且可能出現(xiàn)同一個(gè)元素被同一個(gè)線(xiàn)程先取后放的情況

輕量級(jí)鎖：僅在跨線(xiàn)程竊取元素時(shí)加鎖，核心路徑（同線(xiàn)程存?。o(wú)鎖，性能遠(yuǎn)超全局鎖的 List<T>。

graph TD
    A[線(xiàn)程A] --> A_Queue[本地隊(duì)列A: 1, 3, 5]
    B[線(xiàn)程B] --> B_Queue[本地隊(duì)列B: 2, 4]
    C[線(xiàn)程C] --> C_Queue[本地隊(duì)列C: 6]
    
    D[全局隊(duì)列] --> |工作竊取| A_Queue
    D --> |工作竊取| B_Queue
    D --> |工作竊取| C_Queue

核心 API

核心構(gòu)造函數(shù)

• ConcurrentBag<T>(): 創(chuàng)建空的線(xiàn)程安全集合
• ConcurrentBag<T>(IEnumerable<T>): 用指定集合初始化ConcurrentBag<T>

核心方法 / 屬性

• Add(T item): 向集合添加元素（線(xiàn)程安全），無(wú)返回值
• TryTake(out T result): 嘗試從集合移除并返回任意元素：成功返回true，集合為空返回 false
• Count: 獲取集合中元素的數(shù)量（線(xiàn)程安全，但值為瞬時(shí)快照）
• IsEmpty: 判斷集合是否為空（線(xiàn)程安全，瞬時(shí)快照）
• GetEnumerator(): 返回遍歷集合的枚舉器（遍歷的是瞬時(shí)快照，不保證后續(xù)元素不變）

常用操作

var bag = new ConcurrentBag<string>();

// 插入（極快）
bag.Add("任務(wù)A");
bag.Add("任務(wù)B");

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

// 嘗試偷?。═ryPeek 不存在?。?
if (bag.TryTake(out var stolen)) { /* 處理 */ }

// 計(jì)數(shù)（注意：有一定開(kāi)銷(xiāo)）
int count = bag.Count;

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

// 檢查是否為空
bool isEmpty = bag.IsEmpty;

用法示例

多線(xiàn)程添加與消費(fèi)

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

class ConcurrentBagBasicDemo
{
    static void Main()
    {
        // 創(chuàng)建線(xiàn)程安全的ConcurrentBag
        var bag = new ConcurrentBag<int>();

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

        Console.WriteLine($"\n集合總元素?cái)?shù)：{bag.Count}\n");

        // 2. 多線(xiàn)程消費(fèi)元素（直到集合為空）
        Parallel.For(0, 2, threadId =>
        {
            while (!bag.IsEmpty)
            {
                if (bag.TryTake(out int value))
                {
                    Console.WriteLine($"線(xiàn)程{threadId}：取出 {value}");
                }
                // 避免空循環(huán)占用CPU
                Task.Delay(10).Wait();
            }
        });

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

輸出結(jié)果

線(xiàn)程0：添加 1
線(xiàn)程1：添加 101
線(xiàn)程0：添加 2
線(xiàn)程2：添加 201
...（添加順序無(wú)序）
集合總元素?cái)?shù)：20

線(xiàn)程0：取出 2
線(xiàn)程1：取出 101
線(xiàn)程0：取出 1
線(xiàn)程1：取出 201
...（取出順序≠添加順序，且優(yōu)先取當(dāng)前線(xiàn)程添加的元素）
最終集合是否為空：True

核心現(xiàn)象：

• 添加和取出的順序完全無(wú)序，符合 ConcurrentBag<T> “無(wú)序集合” 的特性；
• 同一線(xiàn)程優(yōu)先取出自己添加的元素（TLS 優(yōu)化的體現(xiàn)）。

并行處理大量獨(dú)立小文件

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

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

對(duì)象池實(shí)現(xiàn)

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);
}

關(guān)鍵特性與適用場(chǎng)景

核心特性

• 順序性: 無(wú)序（Add 順序≠遍歷 / Take 順序）
• 線(xiàn)程安全: 所有操作線(xiàn)程安全，無(wú)需手動(dòng)加鎖
• 性能: 同線(xiàn)程存取：極高（無(wú)鎖）；跨線(xiàn)程竊?。褐校ㄝp量級(jí)鎖）
• 空值支持: 允許添加 null（若T為引用類(lèi)型）
• 遍歷特性: 遍歷的是 “瞬時(shí)快照”，遍歷過(guò)程中集合可修改，不拋出異常
• 容量: 無(wú)固定容量限制，動(dòng)態(tài)擴(kuò)容

最佳適用場(chǎng)景

• 線(xiàn)程自產(chǎn)自銷(xiāo)：線(xiàn)程池線(xiàn)程添加元素后，自己快速取出處理（如線(xiàn)程本地緩存）；
• 無(wú)序批量處理：多線(xiàn)程收集數(shù)據(jù)，無(wú)需保證順序（如日志收集、臨時(shí)數(shù)據(jù)存儲(chǔ)）；
• 低鎖競(jìng)爭(zhēng)場(chǎng)景：大多數(shù)操作由同一線(xiàn)程完成，跨線(xiàn)程操作少。
• 對(duì)象池實(shí)現(xiàn)：重用對(duì)象減少分配
• 并行計(jì)算中間結(jié)果收集
• 生產(chǎn)者即消費(fèi)者模式

不適用場(chǎng)景

• 需要有序存?。喝?FIFO（用 ConcurrentQueue<T> ）、LIFO（用ConcurrentStack<T>）；
• 高跨線(xiàn)程竊?。憾嗑€(xiàn)程頻繁添加，且其他線(xiàn)程頻繁取走（此時(shí)鎖競(jìng)爭(zhēng)多，性能低于ConcurrentQueue<T> ）；
• 索引訪(fǎng)問(wèn)：ConcurrentBag<T> 無(wú)索引（如 bag[0] ），需索引訪(fǎng)問(wèn)用ConcurrentDictionary<TKey, TValue> 或手動(dòng)封裝。

最佳實(shí)踐

優(yōu)先用于生產(chǎn)者-消費(fèi)者同線(xiàn)程場(chǎng)景

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

避免用于生產(chǎn)者-消費(fèi)者分離場(chǎng)景

   // 生產(chǎn)者消費(fèi)者分離
   var sharedBag = new ConcurrentBag<Data>();
   
   // 生產(chǎn)者線(xiàn)程
   Task.Run(() => sharedBag.Add(produce()));
   
   // 消費(fèi)者線(xiàn)程
   Task.Run(() => 
   {
       if (sharedBag.TryTake(out var data))
       {
           consume(data);
       }
   });

總結(jié)

ConcurrentBag<T> 是 .NET 并發(fā)集合中的特殊工具：

• ? 在生產(chǎn)者即消費(fèi)者場(chǎng)景中性能卓越
• ? 內(nèi)置工作竊取機(jī)制
• ? 無(wú)鎖實(shí)現(xiàn)減少競(jìng)爭(zhēng)
• ? 線(xiàn)程本地存儲(chǔ)優(yōu)化

最佳適用場(chǎng)景：

• 線(xiàn)程處理自己生成的任務(wù)
• 對(duì)象池實(shí)現(xiàn)
• 并行計(jì)算的結(jié)果收集
• 工作竊取模式的任務(wù)分發(fā)

到此這篇關(guān)于C#.NET ConcurrentBag<T>設(shè)計(jì)原理與使用場(chǎng)景的文章就介紹到這了,更多相關(guān)C#.NET ConcurrentBag<T>使用內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！