如何优化基于Parallel.ForEachAsync的服务调用代码以提升性能？

我编写了一段调用多个服务并最终将项添加至集合的代码，为提升请求中每个id的处理性能，采用了Parallel.ForEachAsync，但响应耗时超过10秒；若替换为普通foreach，耗时仍达8秒以上。以下是GetUpdates方法代码及ModifiedCollection类定义，希望能得到性能优化的建议：

原代码 #

GetUpdates方法 #

public async Task<ModifiedCollection> GetUpdates(Request request)
{
    ModifiedCollection modifiedCollection = new ModifiedCollection();
    List<string> Ids = request.Ids.Distinct().ToList();
    var options = new ParallelOptions { MaxDegreeOfParallelism = 5 };

    if (Ids.Any())
    {
       await Parallel.ForEachAsync(Ids, options, async (id,token) =>
        {
            var globalId = "";
            var result = await _provider.GetList(id, request.IsActive);

            var baseValue = result.FirstOrDefault();

                if (baseValue?.CustId.FirstOrDefault() == Constants.ABC)
                {
                    var predicate = _provider.GetPredicate(
                                 type: baseValue.type,
                                 id: null);

                    var matter = await _provider.GetItem(request.IsActive, predicate);

                   
                    if (matter != null && matter.CustId.Contains(request.ToRequest.CustId.FirstOrDefault()))
                    {
                        result = await _provider.GetList(matter.MapId, request.IsActive);
                        globalId = id;
                    }
                }

                result = BackDraw(result, request.FromRequest.CustId.FirstOrDefault());

                await Parallel.ForEachAsync(result, options, async (item,token) =>
                {
                    var IsValid = true;
                    if (!String.IsNullOrEmpty(request.ToRequest.EligDate))
                        IsValid = CheckEligibility(item, Convert.ToDateTime(request.ToRequest.EligDate));

                    if (IsValid)
                    {
                        var resultMatter = await _provider.GetStructure(item);
                        var uItem = await GetUItem(resultMatter);

                        if (uItem != null)
                        {
                            if (!uItem.rNotes.IsNullOrEmpty() && !uItem.property.IsNullOrEmpty())
                            {
                                modifiedCollection.ModifiedItem.Add(new ModifiedItem
                                {
                                    Id = item.ObjectId,
                                    Version = int.Parse(item.Version),
                                                                       
                                });
                            }
                        }
                    }
                });
        });
    }
    return modifiedCollection;
}

ModifiedCollection类定义 #

public class ModifiedCollection
{
     public ConcurrentBag<ModifiedItem> ModifiedItemBag { get; set; }
     public Updates()
     {
         UpdatesCollectionBag = new ConcurrentBag<ModifiedItem>();            
     }
}

性能优化建议 #

• 修正线程安全与类定义错误

  • ModifiedCollection类的构造函数名错误，应改为与类名一致：public ModifiedCollection()，同时初始化的字段名要匹配属性名（ModifiedItemBag = new ConcurrentBag<ModifiedItem>()）。
  • 代码中添加元素时误用了modifiedCollection.ModifiedItem.Add，应改为modifiedCollection.ModifiedItemBag.Add，否则会引发未定义行为（属性不存在），且ConcurrentBag才是多线程场景下安全的集合类型。

• 优化并行调度策略

  • 嵌套使用Parallel.ForEachAsync会带来额外的调度开销，内层并行可替换为Task.WhenAll，减少并行框架的调度成本：

    // 替换内层Parallel.ForEachAsync
    var itemTasks = result.Select(async item =>
    {
        // 原item处理逻辑
    });
    await Task.WhenAll(itemTasks);
    

  • 调整MaxDegreeOfParallelism：当前设置为5，可根据下游服务的并发承载能力适当调高（比如10或基于CPU核心数Environment.ProcessorCount），但需避免压垮依赖服务。

• 减少重复计算与冗余操作

  • 提前转换EligDate：将Convert.ToDateTime(request.ToRequest.EligDate)移到外层循环前，避免在每个item处理时重复执行：

    DateTime? eligDate = null;
    if (!string.IsNullOrEmpty(request.ToRequest.EligDate))
        eligDate = DateTime.Parse(request.ToRequest.EligDate);
    

    后续直接使用eligDate进行合法性校验。
  • 检查_provider方法是否支持批量调用：如果GetList、GetItem等方法支持批量参数，将单个id调用改为批量调用，减少网络请求次数（这是最可能的性能瓶颈点）。

• 正确传递取消令牌
  将Parallel.ForEachAsync的token参数传递给所有支持取消的异步方法（如_provider.GetList(id, request.IsActive, token)），确保请求取消时能及时终止不必要的操作，避免无效等待。

• 优化CPU密集型操作
  如果BackDraw、CheckEligibility是CPU密集型方法，可考虑使用并行处理（如Parallel.ForEach）或异步化处理，避免阻塞异步线程。

• 性能监控定位瓶颈
  对_provider的各个服务调用添加耗时监控，明确是网络请求还是本地处理导致的延迟，针对性优化。比如使用Stopwatch记录每个异步调用的耗时，定位最慢的环节。

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