BlockingCollection(T)性能

Question

在我公司工作了一段时间,我们使用了一个自行开发的ObjectPool<T>实现,它提供对其内容的阻止访问.它非常简单:a Queue<T>,a ,object锁定,以及在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.但最近我们发现,由于我们使用无扩展,我们实际上做有Concurrent提供给我们的命名空间(在System.Threading.dll).

当然,我认为既然BlockingCollection<T>是Concurrent命名空间中的核心类之一,它可能会提供比我或我的队友写的更好的性能.

所以我尝试编写一个非常简单的新实现:

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

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

令我惊讶的是,根据一些简单的测试(从多个线程借用/返回池几千次),我们的原始实现在性能方面显着优势BlockingCollection<T>.他们似乎都工作正常 ; 只是我们原来的实现似乎要快得多.

我的问题:

1. 为什么会这样？是因为它BlockingCollection<T>提供了更大的灵活性(我理解它是通过包装来实现的IProducerConsumerCollection<T>),这必然会带来性能开销？
2. 这只是对这个BlockingCollection<T>阶级的错误使用吗？
3. 如果这是一个合适的用途BlockingCollection<T>,我只是没有正确使用？例如,Take/ Add方法是否过于简单化,并且有一种更好的方法来获得相同的功能？

除非有人在回答第三个问题时提供一些见解,否则我们现在似乎将坚持我们原来的实施.

Answer 1

这里有几种可能的可能性.

首先,BlockingCollection<T>Reactive Extensions是一个backport,与.NET 4最终版本不完全相同.如果这个backport的性能不同于.NET 4 RTM,我不会感到惊讶(虽然我没有特别描述这个集合).大多数TPL在.NET 4中的表现都比在.NET 3.5后端表现更好.

话虽这么说,BlockingCollection<T>如果你有一个生产者线程和一个消费者线程,我怀疑你的实现会胜过.对于一个生产者和一个消费者,您的锁对总体性能的影响较小,重置事件是在消费者方面等待的非常有效的方法.

但是,BlockingCollection<T>它旨在允许许多生产者线程非常好地"排队"数据.这对于您的实现来说效果不佳,因为锁定争用将很快变得有问题.

话虽如此,我还想指出一个误解:

......它可能提供比我或我的队友写的更好的表现.

这通常不是真的.框架集合类通常表现很好,但通常不是给定方案的最高性能选项.话虽如此,他们往往表现良好,同时非常灵活和非常强大.它们通常倾向于非常好地扩展."家庭编写的"集合类通常在特定场景中优于框架集合,但在用于特定场景之外的场景中时往往会出现问题.我怀疑这是其中一种情况.

Answer 2

我试图BlockingCollection对一个ConurrentQueue/AutoResetEvent组合是.NET 4(类似于OP的解决方案,但无锁),而后者是康宝这么快很多对我的使用情况下,我抛弃BlockingCollection.不幸的是,差不多一年前,我找不到基准测试结果.

使用单独的AutoResetEvent不会使事情变得复杂得多.事实上,人们甚至可以一劳永逸地把它抽象成BlockingCollectionSlim......

BlockingCollection内部依赖于ConcurrentQueue为好,但确实具有一些附加杂耍苗条信号量和取消标记,这产生额外的特征,但在成本,不使用时也是如此.还应该注意,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);