任务并行库中已排定优先级的队列

因此，这里有一个关于一个相当幼稚的优先级队列的相当幼稚的并发实现。这里的想法是，有一个排序集同时保留真实项目和优先级，但它被赋予了一个只比较优先级的比较器。构造函数采用一个函数来计算给定对象的优先级。

至于实际的实现，它们并没有得到有效的实现，我只是对所有内容进行lock。创建更高效的实现可以防止将SortedSet用作优先级队列，而重新实现其中一个可以同时有效访问的实现并不那么容易。

为了更改项目的优先级，您需要从集合中删除该项目，然后再次添加它，并且要在不迭代整个集合的情况下找到它，您需要知道旧的优先级和新的优先级。

public class ConcurrentPriorityQueue<T> : IProducerConsumerCollection<T>
{
    private object key = new object();
    private SortedSet<Tuple<T, int>> set;
    private Func<T, int> prioritySelector;
    public ConcurrentPriorityQueue(Func<T, int> prioritySelector, IComparer<T> comparer = null)
    {
        this.prioritySelector = prioritySelector;
        set = new SortedSet<Tuple<T, int>>(
            new MyComparer<T>(comparer ?? Comparer<T>.Default));
    }
    private class MyComparer<T> : IComparer<Tuple<T, int>>
    {
        private IComparer<T> comparer;
        public MyComparer(IComparer<T> comparer)
        {
            this.comparer = comparer;
        }
        public int Compare(Tuple<T, int> first, Tuple<T, int> second)
        {
            var returnValue = first.Item2.CompareTo(second.Item2);
            if (returnValue == 0)
                returnValue = comparer.Compare(first.Item1, second.Item1);
            return returnValue;
        }
    }
    public bool TryAdd(T item)
    {
        lock (key)
        {
            return set.Add(Tuple.Create(item, prioritySelector(item)));
        }
    }
    public bool TryTake(out T item)
    {
        lock (key)
        {
            if (set.Count > 0)
            {
                var first = set.First();
                item = first.Item1;
                return set.Remove(first);
            }
            else
            {
                item = default(T);
                return false;
            }
        }
    }
    public bool ChangePriority(T item, int oldPriority, int newPriority)
    {
        lock (key)
        {
            if (set.Remove(Tuple.Create(item, oldPriority)))
            {
                return set.Add(Tuple.Create(item, newPriority));
            }
            else
                return false;
        }
    }
    public bool ChangePriority(T item)
    {
        lock (key)
        {
            var result = set.FirstOrDefault(pair => object.Equals(pair.Item1, item));
            if (object.Equals(result.Item1, item))
            {
                return ChangePriority(item, result.Item2, prioritySelector(item));
            }
            else
            {
                return false;
            }
        }
    }
    public void CopyTo(T[] array, int index)
    {
        lock (key)
        {
            foreach (var item in set.Select(pair => pair.Item1))
            {
                array[index++] = item;
            }
        }
    }
    public T[] ToArray()
    {
        lock (key)
        {
            return set.Select(pair => pair.Item1).ToArray();
        }
    }
    public IEnumerator<T> GetEnumerator()
    {
        return ToArray().AsEnumerable().GetEnumerator();
    }
    IEnumerator IEnumerable.GetEnumerator()
    {
        return GetEnumerator();
    }
    public void CopyTo(Array array, int index)
    {
        lock (key)
        {
            foreach (var item in set.Select(pair => pair.Item1))
            {
                array.SetValue(item, index++);
            }
        }
    }
    public int Count
    {
        get { lock (key) { return set.Count; } }
    }
    public bool IsSynchronized
    {
        get { return true; }
    }
    public object SyncRoot
    {
        get { return key; }
    }
}

一旦您有了IProducerConsumerCollection<T>实例，即上面的对象，您就可以将其用作BlockingCollection<T>的内部支持对象，以获得更易于使用的用户界面。

ParallelExtensionsExtras包含几个自定义TaskScheduler，它们可以直接提供帮助，也可以作为您自己的调度器的基础。

具体来说，有两个调度器可能对您感兴趣：

• QueuedTaskScheduler，它允许您按不同的优先级调度Task，但不允许更改排队的Task的优先级
• ReprioritizableTaskScheduler，它没有不同的优先级，但允许您将特定的Task移动到队列的前面或后面。（尽管更改优先级是当前等待的Task秒数中的O（n），但如果同时有多个Task秒，这可能会成为一个问题。）