C#并行结果合并的效率优化策略，详解线程安全集合、分区收集、归并树三种方案

// 模拟处理电商订单日志
var rawLogs = Enumerable.Range(1, 1000000).Select(i => $"Order_{i}");
var results = new ConcurrentBag<string>(); // 线程安全集合

Parallel.ForEach(rawLogs, log =>
{
    // 模拟复杂处理逻辑：解析日志结构
    var processed = log.Replace("_", "-") + $"_{DateTime.Now:HHmmss}";
    results.Add(processed); // 自动处理并发添加
});

// 最终结果转换
var finalList = results.ToList();

// 处理物联网传感器数据流
const int partitionCount = Environment.ProcessorCount * 2;
var partitions = Partitioner.Create(sensorData, EnumerablePartitionerOptions.NoBuffering);
var localQueues = new ConcurrentQueue<double>[partitionCount];

// 初始化分区容器
for (int i = 0; i < partitionCount; i++)
{
    localQueues[i] = new ConcurrentQueue<double>();
}

Parallel.ForEach(partitions, (data, state, index) =>
{
    // 根据线程索引绑定存储分区
    var targetQueue = localQueues[index % partitionCount];
    
    // 模拟数据清洗：过滤异常值
    if (data > -50 && data < 150) 
    {
        targetQueue.Enqueue(Math.Round(data, 2));
    }
});

// 合并所有分区结果
var finalData = localQueues.SelectMany(q => q).ToList();

// 海量文本特征提取场景
var documents = GetMillionDocuments();
var featureMerger = new FeatureMerger();

Parallel.ForEach(documents.AsParallel().WithMergeOptions(ParallelMergeOptions.AutoBuffered), doc =>
{
    // 提取文本特征向量（假设返回Dictionary）
    var features = ExtractFeatures(doc);
    
    // 使用归并器进行增量合并
    featureMerger.Merge(features);
});

// 自定义归并器实现
class FeatureMerger
{
    private readonly ConcurrentDictionary<string, double> _mergedFeatures = new();
    
    public void Merge(Dictionary<string, double> features)
    {
        foreach (var kv in features)
        {
            _mergedFeatures.AddOrUpdate(kv.Key, kv.Value, 
                (key, oldValue) => oldValue + kv.Value);
        }
    }
}

// 错误示例：导致大量装箱操作
ConcurrentBag resultBag = new ConcurrentBag(); // 非泛型版本
Parallel.For(0, 1000000, i => resultBag.Add(i)); 

// 正确做法：明确指定泛型类型
var properBag = new ConcurrentBag<int>();

// 存在问题的代码模式
Parallel.ForEach(dataSet, item =>
{
    var buffer = new List<byte>(1024); // 每次循环都新建集合
    ProcessItem(item, buffer);
    lock(_locker) { finalList.AddRange(buffer); }
});

// 优化方案：使用对象池
var bufferPool = new ObjectPool<List<byte>>(() => new List<byte>(1024));
Parallel.ForEach(dataSet, item =>
{
    var buffer = bufferPool.Get();
    try {
        ProcessItem(item, buffer);
        lock(_locker) { finalList.AddRange(buffer); }
    }
    finally {
        buffer.Clear();
        bufferPool.Return(buffer);
    }
});

// 时间序列数据处理
var sensorReadings = GetSensorData();

var processed = sensorReadings
    .AsParallel()
    .AsOrdered() // 保持原始顺序
    .WithDegreeOfParallelism(8)
    .Select(r => TransformReading(r))
    .ToList(); // 自动处理有序合并

// 分布式统计计算
var stats = measurements.AsParallel()
    .Aggregate(
        () => new Statistics(), // 初始化局部聚合器
        (localStats, item) => localStats.Accumulate(item), // 局部聚合
        (main, local) => main.Merge(local), // 全局合并
        final => final.GetResult() // 最终格式化
    );

class Statistics
{
    public long Count;
    public double Sum;
    
    public Statistics Accumulate(double value)
    {
        Count++;
        Sum += value;
        return this;
    }
    
    public Statistics Merge(Statistics other)
    {
        Count += other.Count;
        Sum += other.Sum;
        return this;
    }
    
    public double Mean => Sum / Count;
}

// 实验性API（需开启unsafe上下文）
Span<byte> outputBuffer = stackalloc byte[1024];
Parallel.ForEach(partitions, (partition, state) =>
{
    var slice = outputBuffer.Slice(state.Index * sliceSize, sliceSize);
    ProcessPartition(partition, slice);
});