.NET6 中具有优先级队列的 Parallel.ForEach

我正在尝试奔跑Parallel.ForEach on my Priority Queue但我收到以下错误：

严重性代码 说明 项目文件行抑制状态 错误 CS0411 无法从用法推断方法“Parallel.ForEach(OrderablePartitioner, ParallelOptions, Action)”的类型参数。尝试显式指定类型参数。 TPL_POC.PL

我知道如何执行Parallel.ForEach with IEnumerable and List但以下内容没有运气。

private void ProcessTasksParallely()
{
    PriorityQueue<string, int> activeTasksPriority = new PriorityQueue<string, int>();
    foreach (var task in this.tasks)
    {
        activeTasksPriority.Enqueue(task.Task, task.Id);
    }
    Console.WriteLine("Processing");

    var options = new ParallelOptions { MaxDegreeOfParallelism = (Environment.ProcessorCount / 2) * 10 };

    Parallel.ForEach(activeTasksPriority.TryDequeue(out string t, out int priority),
        options,
        (t, priority) =>
        {
            Console.WriteLine($" task {priority}, task = {t}, thread = {Thread.CurrentThread.ManagedThreadId}");
            Thread.Sleep(100);
        });
}

我正在尝试这样做，因为我需要并行处理任务，但要根据它们安排的优先级。

The PriorityQueue<TElement, TPriority> https://learn.microsoft.com/en-us/dotnet/api/system.collections.generic.priorityqueue-2类没有提供将其用作IEnumerable盒子外面。它只有一个UnorderedItems https://learn.microsoft.com/en-us/dotnet/api/system.collections.generic.priorityqueue-2.unordereditems财产，这不是你想要的。此属性产生队列的内容而不消耗它们，并且不按特定顺序。虽然实现自定义很容易GetConsumingEnumerable方法为PriorityQueue<TElement, TPriority>类，像这样：

/// <summary>
/// Gets an enumerable sequence that consumes the elements of the queue
/// in an ordered manner.
/// </summary>
public static IEnumerable<(TElement Element, TPriority Priority)>
    GetConsumingEnumerable<TElement, TPriority>(
    this PriorityQueue<TElement, TPriority> source)
{
    while (source.TryDequeue(out TElement element, out TPriority priority))
    {
        yield return (element, priority);
    }
}

使用示例：

var partitioner = Partitioner.Create(activeTasksPriority.GetConsumingEnumerable(),
    EnumerablePartitionerOptions.NoBuffering);

Parallel.ForEach(partitioner, options, entry =>
{
    var (t, priority) = entry;
    Console.WriteLine($"Priority: {priority}, Task: {t}");
    Thread.Sleep(100);
});

的意图Partitioner.Create https://learn.microsoft.com/en-us/dotnet/api/system.collections.concurrent.partitioner.create+NoBuffering https://learn.microsoft.com/en-us/dotnet/api/system.collections.concurrent.enumerablepartitioneroptions是为了防止Parallel.ForEach在准备好处理元素之前提前消耗元素并将它们存储到缓冲区中。

Note:这个答案涉及问题中提出的简单场景，其中PriorityQueue<E,P>在开始并行循环之前已完全填充。如果您想在循环运行时在队列中添加更多项目，则不能直接使用PriorityQueue<E,P>有两个原因：

1. 它不是线程安全的集合。
2. 它没有阻塞功能，因此循环可能会在处理所有项目之前提前完成。

如果你正在处理这样的场景，你可以看看这个问题：并发优先收集 https://stackoverflow.com/questions/23470196/concurrent-collection-with-priority.