BlockingCollection(T) 性能

Question

在我的公司有一段时间，我们使用了一个自制的 ObjectPool 实现，它提供了对其内容的阻止访问。它非常简单：一个Queue、一个要锁定的对象，以及一个AutoResetEvent，用于在发生“借用”线程时向“借用”线程发出信号。项目已添加。

该类的核心实际上是这两个方法：

public T Borrow() {
    lock (_queueLock) {
        if (_queue.Count > 0)
            return _queue.Dequeue();
    }

    _objectAvailableEvent.WaitOne();

    return Borrow();
}

public void Return(T obj) {
    lock (_queueLock) {
        _queue.Enqueue(obj);
    }

    _objectAvailableEvent.Set();
}

我们一直在使用这个方法和其他一些集合类，而不是 System.Collections.Concurrent 提供的那些方法，因为我们使用的是 .NET 3.5，而不是 4.0。但最近我们发现，由于我们使用响应式扩展，我们实际上< em>确实有可供我们使用的Concurrent命名空间（在System.Threading.dll中）。

当然，我认为自从 BlockingCollection 是 Concurrent 命名空间中的核心类之一，它可能会提供比我或我的队友编写的任何内容更好的性能。

因此，我尝试编写一个工作非常简单的新实现：

public T Borrow() {
    return _blockingCollection.Take();
}

public void Return(T obj) {
    _blockingCollection.Add(obj);
}

令我惊讶的是，根据一些简单的测试（从多个线程借用/返回池几千次），我们最初的实现显着优于BlockingCollectionT>在性能方面。它们似乎都可以正常工作；只是我们原来的实现似乎要快得多。

我的问题：

1. 为什么会这样？可能是因为 BlockingCollection 提供了更大的灵活性（我理解它是通过包装 IProducerConsumerCollection< T>），这必然会带来性能开销吗？
2. 这是否只是对 BlockingCollection 类的完全误导性使用？
3. 如果这是 BlockingCollection 的正确使用，我是否使用不当？例如，Take/Add 方法是否过于简单，是否有一种性能更好的方法来获得相同的功能？

除非有人对第三个问题有一些见解，否则我们现在似乎将坚持原来的实现。

Answer 1

这里有一些潜在的可能性。

首先，响应式扩展中的 BlockingCollection 是向后移植的，与 .NET 4 最终版本并不完全相同。如果此向后移植的性能与 .NET 4 RTM 不同，我不会感到惊讶（尽管我没有具体介绍此集合）。大多数 TPL 在 .NET 4 中的性能比在 .NET 3.5 向后移植中的性能更好。

话虽如此，如果您有一个生产者线程和一个消费者线程，我怀疑您的实现将优于 BlockingCollection。对于一个生产者和一个消费者，你的锁对总体性能的影响会更小，并且重置事件是消费者端等待的一种非常有效的手段。

然而，BlockingCollection 的设计目的是允许许多生产者线程很好地“排队”数据。这对于您的实现来说效果不佳，因为锁定争用很快就会开始出现问题。

话虽这么说，我还想在这里指出一个误解：

...它可能会提供比我或我的队友编写的任何内容更好的性能。

这通常是不正确的。框架集合类通常执行非常好，但对于给定场景通常不是性能最佳的选项。话虽如此，它们往往表现良好，同时非常灵活和强大。它们往往能够很好地扩展。 “自制”集合类在特定场景中通常优于框架集合，但在专门设计的场景之外使用时往往会出现问题。我怀疑这是其中一种情况。

Answer 2

我在 .Net 4 中尝试使用 BlockingCollection 来对抗 ConurrentQueue/AutoResetEvent 组合（类似于 OP 的解决方案，但无锁），而后一个组合如此对于我的用例来说，速度要快得多，所以我放弃了 BlockingCollection。不幸的是，这已经是一年前的事了，我找不到基准测试结果。

使用单独的 AutoResetEvent 不会使事情变得更加复杂。事实上，人们甚至可以将其一劳永逸地抽象为 BlockingCollectionSlim...

BlockingCollection 内部也依赖于 ConcurrentQueue，但是 使用细长信号量和取消标记<进行了一些额外的处理/em>，它会产生额外的功能，但要付出一定的代价，即使不使用也是如此。还应该注意的是，BlockingCollection 并未与 ConcurrentQueue 联姻，但也可以与 IProducerConsumerCollection 的其他实现者一起使用。

---

一个无限制的、非常简单的 BlockingCollectionSlim 实现：

class BlockingCollectionSlim<T>
{
    private readonly ConcurrentQueue<T> _queue = new ConcurrentQueue<T>();
    private readonly AutoResetEvent _autoResetEvent = new AutoResetEvent(false);
    public void Add(T item)
    {
        _queue.Enqueue(item);
        _autoResetEvent.Set();
    }
    public bool TryPeek(out T result)
    {
        return _queue.TryPeek(out result);
    }
    public T Take()
    {
        T item;
        while (!_queue.TryDequeue(out item))
            _autoResetEvent.WaitOne();
        return item;
    }
    public bool TryTake(out T item, TimeSpan patience)
    {
        if (_queue.TryDequeue(out item))
            return true;
        var stopwatch = Stopwatch.StartNew();
        while (stopwatch.Elapsed < patience)
        {
            if (_queue.TryDequeue(out item))
                return true;
            var patienceLeft = (patience - stopwatch.Elapsed);
            if (patienceLeft <= TimeSpan.Zero)
                break;
            else if (patienceLeft < MinWait)
            // otherwise the while loop will degenerate into a busy loop,
            // for the last millisecond before patience runs out
                patienceLeft = MinWait;
            _autoResetEvent.WaitOne(patienceLeft);
        }
        return false;
    }
    private static readonly TimeSpan MinWait = TimeSpan.FromMilliseconds(1);

Answer 3

我在 .Net 4.7.2 中遇到了与 BlockingCollection 相同的性能问题，并找到了这篇文章。我的案例是多个生产者-多个消费者，特别是从多个源读取小数据块，并且应该由多个过滤器处理。使用了多个 (Env.ProcessorCount) BlockingCollections，最终性能分析器告诉我 BlockingCollection.GetConsumingEnumerable.MoveNext() 占用的 CPU 时间比实际过滤更多！

谢谢，@Eugene别列索夫斯基，给你的代码。仅供参考：在我的环境中，它几乎比 BlockingCollection 慢两倍。因此，这是我的 SpinLocked BlockingCollection：

public class BlockingCollectionSpin<T>
{
    private SpinLock _lock = new SpinLock(false);
    private Queue<T> _queue = new Queue<T>();

    public void Add(T item)
    {
        bool gotLock = false;
        try
        {
            _lock.Enter(ref gotLock);
            _queue.Enqueue(item);
        }
        finally
        {
            if (gotLock) _lock.Exit(false);
        }
    }

    public bool TryPeek(out T result)
    {
        bool gotLock = false;
        try
        {
            _lock.Enter(ref gotLock);
            if (_queue.Count > 0)
            {
                result = _queue.Peek();
                return true;
            }
            else
            {
                result = default(T);
                return false;
            }
        }
        finally
        {
            if (gotLock) _lock.Exit(false);
        }
    }

    public T Take()
    {
        var spin = new SpinWait();
        do
        {
            bool gotLock = false;
            try
            {
                _lock.Enter(ref gotLock);
                if (_queue.Count > 0)
                    return _queue.Dequeue();
            }
            finally
            {
                if (gotLock) _lock.Exit(false);
            }
            spin.SpinOnce();
        } while (true);
    }
}

对于性能关键型代码，我建议避免 readonly 字段修饰符。它添加了对 IL 中每个字段访问的检查。使用以下测试代码

private static void TestBlockingCollections()
{
    const int workAmount = 10000000;
    var workerCount = Environment.ProcessorCount * 2;
    var sw = new Stopwatch();
    var source = new long[workAmount];
    var rnd = new Random();
    for (int i = 0; i < workAmount; i++)
        source[i] = rnd.Next(1000000);

    var swOverhead = 0.0;
    for (int i = 0; i < workAmount; i++)
    {
        sw.Restart();
        swOverhead += sw.Elapsed.TotalMilliseconds;
    }
    swOverhead /= workAmount;

    var sum1 = new long[workerCount];
    var queue1 = new BlockingCollection<long>(10000);
    var workers = Enumerable.Range(0, workerCount - 1).Select(n =>
    Task.Factory.StartNew(() =>
    {
        foreach (var l in queue1.GetConsumingEnumerable())
            sum1[n] += l;
    })).ToArray();

    Thread.Sleep(1000);

    sw.Restart();
    foreach (var l in source)
        queue1.Add(l);
    queue1.CompleteAdding();
    Task.WaitAll(workers);
    var elapsed = sw.Elapsed.TotalMilliseconds - swOverhead;
    Console.WriteLine("BlockingCollection {0:F4}ms", elapsed / workAmount);

    var sum2 = new long[workerCount];
    var queue2 = new BlockingCollectionSlim<long?>();
    workers = Enumerable.Range(0, workerCount - 1).Select(n =>
    Task.Factory.StartNew(() =>
    {
        long? l;
        while ((l = queue2.Take()).HasValue)
            sum2[n] += l.Value;
    })).ToArray();

    Thread.Sleep(1000);

    sw.Restart();
    foreach (var l in source)
        queue2.Add(l);
    for (int i = 0; i < workerCount; i++)
        queue2.Add(null);
    Task.WaitAll(workers);
    elapsed = sw.Elapsed.TotalMilliseconds - swOverhead;
    Console.WriteLine("BlockingCollectionSlim {0:F4}ms", elapsed / workAmount);

    var sum3 = new long[workerCount];
    var queue3 = new BlockingCollectionSpin<long?>();
    workers = Enumerable.Range(0, workerCount - 1).Select(n =>
    Task.Factory.StartNew(() =>
    {
        long? l;
        while ((l = queue3.Take()).HasValue)
            sum3[n] += l.Value;
    })).ToArray();

    Thread.Sleep(1000);

    sw.Restart();
    foreach (var l in source)
        queue3.Add(l);
    for (int i = 0; i < workerCount; i++)
        queue3.Add(null);
    Task.WaitAll(workers);
    elapsed = sw.Elapsed.TotalMilliseconds - swOverhead;
    Console.WriteLine("BlockingCollectionSpin {0:F4}ms", elapsed/workAmount);

    if (sum1.Sum() != sum2.Sum() || sum2.Sum() != sum3.Sum())
        Console.WriteLine("Wrong sum in the end!");

    Console.ReadLine();
}

在具有 2 个内核且启用 HT 的 Core i5-3210M 上，我得到以下输出：

BlockingCollection     0.0006ms
BlockingCollectionSlim 0.0010ms (Eugene Beresovsky implementation)
BlockingCollectionSpin 0.0003ms

因此，SpinLocked 版本比 .Net BlockingCollection 快两倍。但是，我建议只使用它！如果您真的更喜欢性能而不是代码简单性（和可维护性）。