C# 线程安全集合 - Concurrent Collections

using System.Collections.Concurrent;

ConcurrentDictionary<string, int> dict = new ConcurrentDictionary<string, int>();

// 添加或更新
dict.TryAdd("apple", 1);
dict.TryAdd("banana", 2);

// 获取值
if (dict.TryGetValue("apple", out int value))
{
    Console.WriteLine($"apple: {value}");
}

// 更新（原子操作）
dict.AddOrUpdate("apple", 
    addValueFactory: key => 1,
    updateValueFactory: (key, oldValue) => oldValue + 1);

// 移除
dict.TryRemove("banana", out int removed);

// 获取或添加
int count = dict.GetOrAdd("orange", 0);

// 遍历时不能直接修改集合，但 ConcurrentDictionary 允许在遍历时修改
// 遍历的是快照，不会抛出异常
foreach (var kvp in dict)
{
    Console.WriteLine($"{kvp.Key}: {kvp.Value}");
}

// 安全修改遍历中的元素
dict.ToList().ForEach(kvp => {
    dict.TryUpdate(kvp.Key, kvp.Value + 10, kvp.Value);
});

ConcurrentQueue<int> queue = new ConcurrentQueue<int>();

// 入队
queue.Enqueue(1);
queue.Enqueue(2);

// 出队
if (queue.TryDequeue(out int result))
{
    Console.WriteLine($"出队: {result}");
}

// 偷看（不移除）
if (queue.TryPeek(out int peek))
{
    Console.WriteLine($"队首: {peek}");
}

// 生产-消费者示例
async Task Producer(ConcurrentQueue<int> q, int items)
{
    for (int i = 0; i < items; i++)
    {
        q.Enqueue(i);
        await Task.Delay(10);
    }
}

async Task Consumer(ConcurrentQueue<int> q, CancellationToken ct)
{
    while (!ct.IsCancellationRequested || q.Count > 0)
    {
        if (q.TryDequeue(out int item))
        {
            Console.WriteLine($"处理: {item}");
        }
        else
        {
            await Task.Delay(50);
        }
    }
}

ConcurrentStack<int> stack = new ConcurrentStack<int>();

// 压栈
stack.Push(1);
stack.PushRange(new[] { 2, 3, 4 });

// 弹栈
if (stack.TryPop(out int result))
{
    Console.WriteLine($"弹出: {result}");
}

// 批量弹出
stack.TryPopRange(out int[] items, 3);

ConcurrentBag<string> bag = new ConcurrentBag<string>();

// 添加
bag.Add("task1");
bag.Add("task2");

// 取出
if (bag.TryTake(out string item))
{
    Console.WriteLine($"取出: {item}");
}

// 偷看
if (bag.TryPeek(out string peek))
{
    Console.WriteLine($"下一个: {peek}");
}

ConcurrentBag<int> bag = new ConcurrentBag<int>();

Parallel.For(0, 1000, i => {
    bag.Add(i);  // 每个线程添加到自己的本地存储
});

Console.WriteLine($"总数: {bag.Count}");  // 1000

// 取出时优先从本地队列取，如果没有再偷取其他线程的
int sum = 0;
while (bag.TryTake(out int item))
{
    sum += item;
}

using System.Collections.Concurrent;

// 默认包装 ConcurrentQueue（FIFO）
BlockingCollection<int> collection = new BlockingCollection<int>();

// 生产者
Task producer = Task.Run(() => {
    for (int i = 0; i < 10; i++)
    {
        collection.Add(i);
        Console.WriteLine($"生产: {i}");
        Thread.Sleep(100);
    }
    collection.CompleteAdding();  // 标记不再添加
});

// 消费者
Task consumer = Task.Run(() => {
    foreach (var item in collection.GetConsumingEnumerable())
    {
        Console.WriteLine($"消费: {item}");
    }
});

await Task.WhenAll(producer, consumer);

// 容量为 10，超过会阻塞生产者
BlockingCollection<int> bounded = new BlockingCollection<int>(10);

// 使用 ConcurrentBag
BlockingCollection<string> bagCollection = new BlockingCollection<string>(
    new ConcurrentBag<string>(), 
    boundedCapacity: 20);

if (collection.TryAdd(42, TimeSpan.FromSeconds(1)))
{
    Console.WriteLine("添加成功");
}

if (collection.TryTake(out int item, TimeSpan.FromSeconds(1)))
{
    Console.WriteLine($"取出: {item}");
}

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

class PipelineExample
{
    static async Task Main()
    {
        var stage1 = new BlockingCollection<string>(10);
        var stage2 = new BlockingCollection<(string, int)>(10);
        
        // 阶段1：读取数据
        var reader = Task.Run(() => {
            for (int i = 1; i <= 100; i++)
            {
                stage1.Add($"item_{i}");
            }
            stage1.CompleteAdding();
        });
        
        // 阶段2：转换处理
        var processor = Task.Run(() => {
            foreach (var item in stage1.GetConsumingEnumerable())
            {
                int length = item.Length;
                stage2.Add((item, length));
            }
            stage2.CompleteAdding();
        });
        
        // 阶段3：输出
        foreach (var (item, length) in stage2.GetConsumingEnumerable())
        {
            Console.WriteLine($"{item}: {length} chars");
        }
        
        await Task.WhenAll(reader, processor);
    }
}

// 不推荐：在遍历时进行大量操作
foreach (var item in concurrentBag)
{
    // 每个操作都涉及原子性的开销
}

// 推荐：先取出再处理
while (concurrentBag.TryTake(out var item))
{
    Process(item);
}