基于Parallel类的线程安全进度计数器性能优化咨询

你的问题很典型——在并行任务中既要保证进度统计的线程安全，又不想因为锁导致性能暴跌。先分析下你现有代码的问题：你把整个进度计算和日志逻辑都放在锁里，而且每次循环迭代都要进入锁，这会导致大量线程阻塞，性能自然上不去；而去掉锁后，共享的progress和last变量在多线程下并发读写，就会出现重复日志或进度计算错误。

下面给你几个优化方向，一步步解决问题：

1. 用原子操作替代锁更新进度计数 #

首先，进度计数的更新完全不需要锁——.NET提供了Interlocked类的原子操作，能在无锁的情况下安全地递增共享计数器，性能比锁高得多。

2. 缩小锁的粒度，只在必要时加锁 #

日志输出是唯一需要同步的环节（避免重复打印相同进度），所以我们只在要输出日志的时候才加锁，而且用双重检查来进一步减少锁竞争。

优化后的代码示例 #

// 线程安全的完成任务计数器
int completedTasks = 0;
// 记录上一次输出的进度百分比，避免重复日志
int lastReportedProgress = -1;
// 仅用于同步日志输出的锁对象
object logSyncLock = new object();

Parallel.ForEach(bossUsers, StandardParallelOptions(), async (user) =>
{
    logger.Trace($"Sync new user {user.PID}.");

    // 👉 这里可以优化：用SemaphoreSlim替代队列+Thread.Sleep的轮询，效率更高
    // 先保留你的原有逻辑，后面再给你优化方案
    while (queue.Count >= MaxDegreeOfParallelism)
    {
        Thread.Sleep(1);
    }
    queue.Enqueue(user);

    try
    {
        await CreateOrUpdateUserAsync(cardRepository, dbScope, user, logger, cancellationToken)
            .ConfigureAwait(false);
    }
    finally
    {
        // 原子递增完成数，无锁线程安全
        int currentCompleted = Interlocked.Increment(ref completedTasks);
        // 计算当前进度百分比
        int currentProgress = (int)(((decimal)currentCompleted / bossUsers.Count) * 100);

        // 只有当进度是10的倍数，且和上一次报告的进度不同时，才触发日志
        if (currentProgress % 10 == 0 && currentProgress != lastReportedProgress)
        {
            // 仅在需要输出日志时加锁，锁的粒度极小
            lock (logSyncLock)
            {
                // 双重检查：防止多个线程同时到达这里，重复输出相同进度
                if (currentProgress != lastReportedProgress)
                {
                    logger.Info($"Progress: {currentProgress}%");
                    lastReportedProgress = currentProgress;
                }
            }
        }
    }

    logger.Trace($"Sync user {user.PID} complete.");
});

额外优化：替换低效的队列轮询逻辑 #

你原来用queue.Count+Thread.Sleep(1)来控制并发度，这种轮询方式会浪费线程资源。更高效的方式是用SemaphoreSlim，它可以让线程在等待时释放CPU资源，而不是空转：

// 初始化信号量，最大并发数等于你的MaxDegreeOfParallelism
using var concurrencySemaphore = new SemaphoreSlim(MaxDegreeOfParallelism, MaxDegreeOfParallelism);
// 线程安全的完成任务计数器
int completedTasks = 0;
// 记录上一次输出的进度百分比，避免重复日志
int lastReportedProgress = -1;
// 仅用于同步日志输出的锁对象
object logSyncLock = new object();

Parallel.ForEach(bossUsers, StandardParallelOptions(), async (user) =>
{
    logger.Trace($"Sync new user {user.PID}.");

    // 等待信号量，获取并发许可
    await concurrencySemaphore.WaitAsync(cancellationToken);
    
    try
    {
        await CreateOrUpdateUserAsync(cardRepository, dbScope, user, logger, cancellationToken)
            .ConfigureAwait(false);
        
        // 进度统计逻辑
        int currentCompleted = Interlocked.Increment(ref completedTasks);
        int currentProgress = (int)(((decimal)currentCompleted / bossUsers.Count) * 100);

        if (currentProgress % 10 == 0 && currentProgress != lastReportedProgress)
        {
            lock (logSyncLock)
            {
                if (currentProgress != lastReportedProgress)
                {
                    logger.Info($"Progress: {currentProgress}%");
                    lastReportedProgress = currentProgress;
                }
            }
        }
    }
    finally
    {
        // 释放信号量许可，让其他线程可以执行
        concurrencySemaphore.Release();
    }

    logger.Trace($"Sync user {user.PID} complete.");
});

为什么这样优化有效？ #

• 原子操作更新计数：Interlocked.Increment是CPU级别的原子指令，不需要线程阻塞，性能几乎和单线程一样。
• 极小的锁粒度：锁只在需要输出日志时才被获取，而且持有时间极短（只做日志输出和更新lastReportedProgress），锁竞争的概率极低。
• 双重检查避免重复日志：即使多个线程同时计算出相同的进度，也只有第一个进入锁的线程会输出日志，后续线程会发现lastReportedProgress已经更新，直接跳过。

这样既保证了进度统计的准确性，又不会对并行任务的性能造成明显影响。

内容的提问来源于stack exchange，提问作者Евгений