使用Parallel.Foreach改造CRM导入代码遇多线程问题求助

嘿，作为多线程编程新手，踩Parallel.Foreach的坑太正常啦！我来帮你理清楚怎么解决这个问题——核心要解决两个关键点：OrganizationRequestCollection的线程安全性，以及锁的正确使用位置。

首先明确：OrganizationRequestCollection不是线程安全的 #

OrganizationRequestCollection是Dynamics CRM SDK中的集合类，它内部的增删改查操作没有内置线程安全机制。如果多个线程同时对同一个实例执行Add、Remove或者遍历操作，必然会出现异常（比如集合计数错误、抛出InvalidOperationException）或者数据错乱的情况。

两种正确改造方案 #

方案1：每个线程创建本地集合，最后合并（推荐，性能更好） #

这种方式能避免锁的使用，减少线程竞争，是最优解。思路是让每个并行线程独立处理自己的请求集合，最后把所有线程的结果合并到全局集合中。

public void ImportDataToCRM(List<Invoice> invoicesList, List<InvoiceLine> invoiceLinesList)
{
    // 用线程安全的ConcurrentBag来存储每个线程的本地请求集合
    var threadLocalCollections = new ConcurrentBag<OrganizationRequestCollection>();

    Parallel.ForEach(invoicesList, invoice =>
    {
        // 每个线程创建自己的OrganizationRequestCollection实例
        var localRequests = new OrganizationRequestCollection();
        
        // 1. 处理当前Invoice，生成对应的CRM请求
        var createInvoiceReq = new CreateRequest 
        { 
            Target = MapInvoiceToCrmEntity(invoice) // 假设这是你的实体映射方法
        };
        localRequests.Add(createInvoiceReq);
        
        // 2. 处理关联的InvoiceLine，生成对应的请求
        var relatedLines = invoiceLinesList.Where(line => line.InvoiceId == invoice.Id).ToList();
        foreach (var line in relatedLines)
        {
            var createLineReq = new CreateRequest 
            { 
                Target = MapInvoiceLineToCrmEntity(line) 
            };
            localRequests.Add(createLineReq);
        }
        
        // 将本地集合加入线程安全的ConcurrentBag
        threadLocalCollections.Add(localRequests);
    });

    // 合并所有线程的请求到全局集合
    var allRequests = new OrganizationRequestCollection();
    foreach (var localColl in threadLocalCollections)
    {
        foreach (var req in localColl)
        {
            allRequests.Add(req);
        }
    }

    // 批量提交到CRM
    using (var crmService = new CrmServiceClient(yourConnectionString))
    {
        crmService.ExecuteMultiple(allRequests);
    }
}

方案2：共享集合+锁（仅当必须共享时使用） #

如果业务逻辑要求必须使用同一个全局OrganizationRequestCollection，那就要在修改共享集合的代码块上加锁，并且尽量缩小锁的范围（只在必要时加锁，避免长时间占用锁影响性能）。

public void ImportDataToCRM(List<Invoice> invoicesList, List<InvoiceLine> invoiceLinesList)
{
    var allRequests = new OrganizationRequestCollection();
    // 创建一个专门用于锁的对象（必须是所有线程共享的引用类型）
    var lockObject = new object();

    Parallel.ForEach(invoicesList, invoice =>
    {
        // 先在本地处理所有请求生成逻辑，避免在锁内做耗时操作
        var localRequests = new List<OrganizationRequest>();
        
        var createInvoiceReq = new CreateRequest 
        { 
            Target = MapInvoiceToCrmEntity(invoice) 
        };
        localRequests.Add(createInvoiceReq);
        
        var relatedLines = invoiceLinesList.Where(line => line.InvoiceId == invoice.Id).ToList();
        foreach (var line in relatedLines)
        {
            var createLineReq = new CreateRequest 
            { 
                Target = MapInvoiceLineToCrmEntity(line) 
            };
            localRequests.Add(createLineReq);
        }

        // 只有往共享集合添加数据时才加锁，且只加锁这一段代码
        lock (lockObject)
        {
            foreach (var req in localRequests)
            {
                allRequests.Add(req);
            }
        }
    });

    // 批量提交到CRM
    using (var crmService = new CrmServiceClient(yourConnectionString))
    {
        crmService.ExecuteMultiple(allRequests);
    }
}

额外注意事项 #

• 检查映射方法的线程安全性：比如MapInvoiceToCrmEntity和MapInvoiceLineToCrmEntity，如果这些方法使用了全局变量、静态缓存或者其他共享资源，也要确保它们是线程安全的，必要时加锁保护。
• 控制并行度：如果invoicesList数据量很大，可以通过ParallelOptions限制最大并行数，避免给CRM服务造成过大压力：

  var parallelOptions = new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount * 2 };
  Parallel.ForEach(invoicesList, parallelOptions, invoice => { ... });
  

• 只读集合的安全：invoiceLinesList如果在并行过程中不会被修改，那么Where遍历是安全的；如果有其他线程同时修改这个列表，也要给它加锁保护。

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