.net - 具有有序输入的 Parallel.ForEach？

Question

我的程序中使用了 Parallel.ForEach() 语句。它使用一些对象的列表作为输入。我不关心输出顺序，但我需要这个循环以与输入列表中相同的顺序获取输入元素。是否可以使用 Parallel.ForEach() 实现此目的？

Answer 1

如果您需要保留来自 IEnumerable<T> 的订单, 你可能想要 implement a custom partitioner的 OrderablePartitioner<T> 种类。此类的示例代码包括一个简单示例，该示例以递增顺序从枚举中一次检索一个。

然而，对于一些可能是简单的生产者-消费者模型的东西来说，这是很多工作，比如 ConcurrentQueue<T> 。 :

var queue = new ConcurrentQueue<X>(yourEnumerableOfX);
Action consumer = () =>
{
    X x;
    while (queue.TryDequeue(out x))
    {
        x.Frob();
    }
};

// At most N "in flight"
int maxParallelism = Environment.ProcessorCount;
var consumers = Enumerable.Repeat(consumer, maxParallelism).ToArray();
Parallel.Invoke(consumers);

使用此代码，您将保证先进先出的行为，并且您的请求最终几乎按照收到的顺序“进行中”处理。一旦并排放置，您将无法保证它们保持顺序。

或者，您可以使用以下方法(限制队列项目的数量保持固定):

// Executes exactly queue.Count iterations at the time of Parallel.ForEach
// due to "snapshot" isolation of ConcurrentQueue<X>.GetEnumerator()
var queue = new ConcurrentQueue<X>(yourEnumerableOfX);
Parallel.ForEach(
    queue,
    _ =>
    {
        X x;
        if (queue.TryDequeue(out x))
        {
            x.Frob();
        }
    });

如果您想继续在一个线程中生产，并在其他线程中消费，请使用 BlockingCollection<T> 以队列作为其后备集合:

var queue = new BlockingCollection<X>(new ConcurrentQueue<X>());

// add to it
Task.Factory.StartNew( () =>
    {
         foreach (var x in yourEnumerableOfX)
         {
             queue.Add(x);
             Thread.Sleep(200);
         }

         // Signal to our consumers we're done:
         queue.CompleteAdding();
    });

现在我们需要“无界”消费者，因为我们不确定究竟有多少队列项目可能存在:

// Roughly the same consumer code as above, but 'unbounded'
Action consumer = () =>
{
    while (!queue.IsCompleted)
    {
        X x;
        try
        {
            // blocking form, switch to TryTake and maybe Thread.Sleep()
            x = queue.Take();
        }
        catch (InvalidOperationException)
        {
            // none left
            break;
        }

        x.Frob();
    }
};

int maxParallelism = Environment.ProcessorCount;
var consumers = Enumerable.Repeat(consumer, maxParallelism).ToArray();
Parallel.Invoke(consumers);