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...
}

根据这篇文章,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();
   

   即使您WithExecutionMode(ParallelExecutionMode.ForceParallelism)在单个核心/ CPU盒上强制执行并行操作,或者当并行度为1时,您的设置也不起作用,因为无法实现真正​​的并行性.

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#代码并非并行运行,而实际上是这样.