C# Parallel.ForEach服务调用的性能与线程问题咨询

数据老陈

2026-5-7

咱们来逐个拆解你的问题，结合代码细节给出针对性的分析和解决方案：

问题1：服务器性能与线程数限制

首先要明确：你的场景是IO密集型操作（调用邮件服务属于网络IO，CPU占用极低），而Parallel.ForEach默认是根据CPU核心数来调度线程的，这个策略并不适配IO密集型场景。如果不限制线程数，处理1000条邮件时可能会出现两个核心问题：

服务器线程池资源耗尽：过多的线程会导致上下文切换开销剧增，反而拖慢整体性能，甚至影响服务器上的其他应用。
邮件服务限流/拒绝：多数邮件服务都有并发请求限制，短时间内发起上千个请求很可能被限流，反而降低处理效率。

解决方案：手动设置MaxDegreeOfParallelism来限制并发数，建议根据服务器配置和邮件服务的并发能力调整（比如8核服务器设16，4核设8，或者参考邮件服务的并发上限）。修改你的并行调用代码：

Parallel.ForEach(orderList, new ParallelOptions { MaxDegreeOfParallelism = 16 }, (orderItem) => {
    // 原有的SendNotifs和集合添加逻辑
});

这样既能保留并行带来的性能提升，又不会过度消耗服务器资源或触发邮件服务的限制。

问题2：集合线程安全与Thread.Sleep的无效性

你遇到的“服务返回邮箱但更新时不存在”的问题，根本原因是List<notify>不是线程安全的集合！Parallel.ForEach是多线程执行，多个线程同时调用notifList.Add()时，会导致集合内部的计数、数组扩容等操作出现竞态条件，最终表现为元素丢失、索引错乱甚至程序崩溃。

而Thread.Sleep(1000)完全无法解决这个问题——它只是让主线程等待，并行线程的并发修改问题依然存在，甚至可能因为等待时间不够还是会出现数据丢失。

正确的两种解决方案：

使用线程安全的集合ConcurrentBag<notify>（适合这种无序添加的场景）：

// 把List换成ConcurrentBag
ConcurrentBag<notify> notifList = new ConcurrentBag<notify>();

// Parallel.ForEach里直接Add，无需额外锁
Parallel.ForEach(..., (orderItem) => {
    SendNotifs(orderItem);
    notifList.Add(new notify() { /* 初始化逻辑 */ });
});

如果需要保持集合有序（你的分页逻辑看起来不依赖顺序），可以用lock包裹Add操作：

List<notify> notifList = new List<notify>();
object lockObj = new object(); // 单独的锁对象，避免锁List本身

Parallel.ForEach(..., (orderItem) => {
    SendNotifs(orderItem);
    var newNotify = new notify() { /* 初始化逻辑 */ };
    lock(lockObj) {
        notifList.Add(newNotify);
    }
});

推荐第一种方案，ConcurrentBag是专门为并行场景设计的，性能更优。

问题3：异常时的线程取消处理

是的，出现异常时需要显式处理线程取消，否则部分并行线程可能继续执行，造成资源浪费或数据不一致。

当前你的代码有两个明显的问题：

只捕获了ApplicationException，但Parallel.ForEach中抛出的异常会被包装成AggregateException，如果不处理，会直接终止程序。
没有使用取消令牌，当异常发生时，无法通知正在运行的并行线程停止工作。

优化方案：使用CancellationTokenSource来控制并行任务的取消，同时正确处理AggregateException：

public void notificationMethod() {
    ConcurrentBag<notify> notifList = new ConcurrentBag<notify>();
    List<orders> orderList = GetNotifs();
    using var cts = new CancellationTokenSource(); // 声明取消令牌源
    try {
        if (orderList.Count > 0) {
            Parallel.ForEach(orderList, new ParallelOptions { 
                MaxDegreeOfParallelism = 16,
                CancellationToken = cts.Token // 传入取消令牌
            }, (orderItem, state) => {
                // 检查是否已取消，提前退出
                if (cts.Token.IsCancellationRequested) {
                    state.Stop();
                    return;
                }
                try {
                    SendNotifs(orderItem);
                    notifList.Add(new notify() { /* 初始化逻辑 */ });
                } catch (Exception ex) {
                    // 捕获单个任务的异常，触发全局取消
                    cts.Cancel();
                    throw; // 重新抛出，让Parallel包装成AggregateException
                }
            });
            // 后续的分页更新逻辑...
        }
    } catch (AggregateException ex) {
        // 逐个处理每个并行任务抛出的异常
        foreach (var innerEx in ex.InnerExceptions) {
            // 记录日志或做其他异常处理
        }
    } catch (ApplicationException ex) {
        // 原有的ApplicationException处理逻辑
        cts.Cancel(); // 触发取消，终止剩余并行任务
    }
}

这样，当任何一个并行任务抛出异常时，会立即触发取消令牌，通知其他正在运行的任务停止工作，避免无效的资源消耗。同时，通过捕获AggregateException，可以逐个处理每个任务的异常，方便排查问题。

内容的提问来源于stack exchange，提问作者Priya Payyavula