C#高效实现并行与并发的最佳实践

// 矩阵乘法加速（使用 SIMD 指令）
void MultiplyMatrices(float[,] matA, float[,] matB, float[,] result)
{
    int size = matA.GetLength(0);
    
    // 使用硬并行度 (物理核心数)
    var parallelOptions = new ParallelOptions 
    { 
        MaxDegreeOfParallelism = Environment.ProcessorCount 
    };
    
    Parallel.For(0, size, parallelOptions, i => 
    {
        for (int j = 0; j < size; j++)
        {
            Vector<float> sum = Vector<float>.Zero;
            for (int k = 0; k < size; k += Vector<float>.Count)
            {
                Vector<float> aVec = new Vector<float>(matA, i, k);
                Vector<float> bVec = new Vector<float>(matB, k, j);
                sum += aVec * bVec;  // SIMD 并行计算
            }
            result[i, j] = Vector.Dot(sum, Vector<float>.One);
        }
    });
}

// 多源数据聚合计算
async Task ProcessMultiSourceAsync()
{
    var source1Task = FetchDataFromAPI("https://source1.com");
    var source2Task = LoadDatabaseDataAsync("DataSource=...");
    var source3Task = ReadLocalFileAsync("data.json");
    
    // 并行等待所有任务
    await Task.WhenAll(source1Task, source2Task, source3Task);
    
    // 安全合并结果（避免锁机制）
    var results = new [] { 
        source1Task.Result, 
        source2Task.Result, 
        source3Task.Result 
    };
    
    var finalResult = CombineData(results);
}

// 高性能并发通道
async Task RunConcurrentPipelineAsync()
{
    // 优化选项：减少内存分配
    var options = new UnboundedChannelOptions
    {
        AllowSynchronousContinuations = false,
        SingleReader = false,  // 支持多消费者
        SingleWriter = false   // 支持多生产者
    };
    
    var channel = Channel.CreateUnbounded<DataItem>(options);
    var producerTasks = new List<Task>();
    
    // 启动 3 个生产者
    for (int i = 0; i < 3; i++)
    {
        producerTasks.Add(ProduceItemsAsync(channel.Writer));
    }
    
    // 启动 4 个消费者
    var consumerTasks = Enumerable.Range(1, 4)
        .Select(_ => ConsumeItemsAsync(channel.Reader))
        .ToArray();
    
    // 等待生产完成
    await Task.WhenAll(producerTasks);
    channel.Writer.Complete();
    
    // 等待消费完成
    await Task.WhenAll(consumerTasks);
}

// 分页数据的并发批处理 (.NET 6+)
async Task BatchProcessAsync(IEnumerable<int> allItems)
{
    // 使用 Parallel.ForEachAsync 限流
    await Parallel.ForEachAsync(
        source: allItems,
        parallelOptions: new ParallelOptions 
        { 
            MaxDegreeOfParallelism = 10,  // 限制并发度
            CancellationToken = _cts.Token
        },
        async (item, ct) => 
        {
            await using var semaphore = new SemaphoreSlimDisposable(5); // 细粒度控制
            await semaphore.WaitAsync(ct);
            try
            {
                await ProcessItemAsync(item, ct);
            }
            finally
            {
                semaphore.Release();
            }
        });
}
 
// 自动释放的信号量包装器
struct SemaphoreSlimDisposable : IAsyncDisposable
{
    private readonly SemaphoreSlim _semaphore;
    public SemaphoreSlimDisposable(int count) => _semaphore = new SemaphoreSlim(count);
    public ValueTask WaitAsync(CancellationToken ct) => _semaphore.WaitAsync(ct).AsValueTask();
    public void Release() => _semaphore.Release();
    public ValueTask DisposeAsync() => _semaphore.DisposeAsync();
}

// 避免伪共享（False Sharing）
class FalseSharingSolution
{
    [StructLayout(LayoutKind.Explicit, Size = 128)]
    struct PaddedCounter
    {
        [FieldOffset(64)] // 每个计数器独占缓存行
        public long Counter;
    }
    
    private readonly PaddedCounter[] _counters = new PaddedCounter[4];
    
    public void Increment(int index) => Interlocked.Increment(ref _counters[index].Counter);
}

// 为高优先级任务创建专用线程池
static TaskFactory HighPriorityTaskFactory
{
    get
    {
        var threadCount = Environment.ProcessorCount / 2;
        var threads = new Thread[threadCount];
        
        for (int i = 0; i < threadCount; i++)
        {
            var t = new Thread(() => Thread.CurrentThread.Priority = ThreadPriority.Highest)
            {
                IsBackground = true,
                Priority = ThreadPriority.Highest
            };
            threads[i] = t;
        }
        
        var taskScheduler = new ConcurrentExclusiveSchedulerPair(
            TaskScheduler.Default, threadCount).ConcurrentScheduler;
            
        return new TaskFactory(CancellationToken.None, 
            TaskCreationOptions.DenyChildAttach,
            TaskContinuationOptions.None,
            taskScheduler);
    }
}
 
// 使用示例
HighPriorityTaskFactory.StartNew(() => ExecuteCriticalTask());

// 使用 ConcurrencyVisualizer
async Task TrackParallelism()
{
    using (var listener = new ConcurrencyVisualizerTelemetry())
    {
        // 标记并行区域
        listener.BeginOperation("Parallel_Core");
        await ProcessBatchParallelAsync();
        listener.EndOperation("Parallel_Core");
        
        // 标记串行区域
        listener.BeginOperation("Sync_Operation");
        RunSyncCalculation();
        listener.EndOperation("Sync_Operation");
    }
}

# 查看线程池使用情况
dotnet-counters monitor -p PID System.Threading.ThreadPool
 
# 检测锁竞争
dotnet-dump collect --type Hang -p PID