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