c# - AsParallel() 顺序执行

Question

我有以下 PLINQ 查询:

// Let's get a few customers
List<Customer> customers = CustomerRepository.GetSomeCustomers();

// Let's get all of the items for all of these customers
List<CustomerItem> items = customers
    .AsParallel()
    .SelectMany(x => ItemRepository.GetItemsByCustomer(x))
    .ToList();

我希望 GetItemsByCustomer() 为每个客户并行执行，但它按顺序运行。

我已经尝试过强制并行，但仍然没有成功:

List<CustomerItem> items = customers
    .AsParallel()
    .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
    .SelectMany(x => ItemRepository.GetItemsByCustomer(x))
    .ToList();

方法签名:

private IEnumerable<Item> GetItemsByCustomer(Customer customer)
{
    // Get all items for a customer...
}

根据 this article ，如果 PLINQ 认为合适，当然可以采用顺序路线，但强制并行性仍应覆盖这一点。

注意:上面的示例纯粹是说明性的 - 假设 customers 是一个小列表，GetItemsByCustomer 是一个昂贵的方法。

Answer 1

AsParallel() 没有任何问题。如果可能，它将并行运行，并且您的 LINQ 表达式中没有顺序依赖性，因此没有任何强制它顺序运行的东西。

您的代码不能并行运行的几个原因可能是:

1. 您的 box/vm 有一个 CPU，或者您有一个 .NET 设置来将并行度限制为一个 CPU。您可以使用以下代码模拟它:

         var customers = new List<Customer>() { new Customer() {Name = "Mick", Surname = "Jagger"}, new Customer() {Name = "George", Surname = "Clooney"},new Customer() {Name = "Kirk", Surname = "DOuglas"}};
   
         var items = customers
           .AsParallel()
           .SelectMany(x =>
           {
                Console.WriteLine("Requesting: " + x.Name + " - " + DateTime.Now);
                Thread.Sleep(3000);
                return new List<CustomerItem>();
   
           })
           .WithDegreeOfParallelism(1)
           .ToList();
   

   即使您在单个核心/CPU 盒上或当并行度为 1 时使用 WithExecutionMode(ParallelExecutionMode.ForceParallelism) 强制并行化，您的设置也不会生效，因为真正的并行化不是可能。

2. 您的存储库中的共享资源发生了一些线程锁定。您可以使用以下代码模拟线程锁定:

       var customers = new List<Customer>() { new Customer() {Name = "Mick", Surname = "Jagger"}, new Customer() {Name = "George", Surname = "Clooney"},new Customer() {Name = "Kirk", Surname = "DOuglas"}};
   
       var locker = new object();
   
       // Let's get all of the items for all of these customers
       var items = customers
           .AsParallel()
           .SelectMany(x =>
           {
               lock (locker)
               {
                   Console.WriteLine("Requesting: " + x.Name + " - " + DateTime.Now);
                   Thread.Sleep(3000);
                   return new List<CustomerItem>();
               }
   
           })
           .ToList();
   

3. 某些数据库设置强制查询/读取在某些情况下按顺序进行，这可能会给您一种印象，即您的 C# 代码没有并行运行，而实际上是这样。