复查请求:线程安全队列(并行推送POP)

它看起来是一个很好的实现。使用不同的锁对我来说总是一个危险信号,所以我仔细看了一眼 一些 涉及同时调用 Pop 和 Push 看起来很安全。我怀疑你可能是在链表上自学的,这意味着一个阻塞队列?这是安全的原因是因为你只参考过 LastNode 从 推 和 FirstNode 从 流行音乐 否则整个骗局就会失败。

现在唯一让我担心的是当你试图从一个 Semaphore 如果它已经满了,它将抛出一个异常,因此您可能希望防范它。 ^一 否则,您将在链接列表中得到额外的节点,并且队列将1)具有超过最大项目数的项目,2)它将被实时锁定。

更新:

我再考虑一下。那 Release 呼叫 推 方法是 非常 有问题。我只看到一个可能的解决方案。删除 maxItems 来自构造函数的参数,并允许信号量计数为 Int.MaxValue . 否则,您将不得不从根本上改变您的方法,从而导致一个与您当前拥有的方法不太接近的实现。

^一 _{我想你会发现这比你马上意识到的更困难。}

1. 考虑添加第二个节点构造函数:

        public Node(T value)
        {
            Value = value;
        }

然后您的客户代码:

            Node newNode = new Node();
            newNode.Value = value;

可以将该值视为不变量:

            Node newNode = new Node(value);

2. 然后让你的公共领域:

        public T Value;
        public Node NextNode;

自动属性:

        public T Value { get; private set; };
        public Node NextNode { get; set; };

因此,您可以从实现中抽象出用法,并在事实发生后添加验证、其他处理等,而对客户机代码的干扰最小。

如果你这样做是为了自我教育,很好-否则 BlockingCollection<T> 或 ConcurrentQueue<T> 是很好的选择。

我在这里看到的一个问题是无法中断 Pop 一旦它启动——它假设一个对象在它醒来时正在等待。在终止合同时,你将如何解决这个问题?一 TryPop 它会回来的 true (有元素)或 false (如果没有数据)可能更好,那么您可以发出信号,等待线程在队列耗尽后立即干净地关闭。

如果你有.NET 4,请看这个

• http://blogs.msdn.com/b/csharpfaq/archive/2010/08/12/blocking-collection-and-the-producer-consumer-problem.aspx

因为我喜欢不可变的物体,这里有一个替代我之前的答案,我会考虑更干净一点:

public sealed class CustomQueue<T> where T : class
{
    private readonly object pushLocker = new object();
    private readonly object popLocker = new object();
    private readonly Semaphore queueSemaphore;
    private readonly int maxItems;
    private volatile int pushIncrement;
    private volatile int popIncrement;
    private Node firstNode = new Node();
    private Node lastNode;

    public CustomQueue(int maxItems)
    {
        this.maxItems = maxItems;
        this.lastNode = this.firstNode;
        this.queueSemaphore = new Semaphore(0, this.maxItems);
    }

    public int Size
    {
        get
        {
            return this.pushIncrement - this.popIncrement;
        }
    }

    public bool Push(T value)
    {
        lock (this.pushLocker)
        {
            if (this.Size >= this.maxItems)
            {
                lock (this.popLocker)
                {
                    this.pushIncrement = this.pushIncrement - this.popIncrement;
                    this.popIncrement = 0;
                    return false;
                }
            }

            Node newNode = new Node(value, this.lastNode.NextNode);

            this.lastNode = new Node(this.lastNode.Value, newNode);
            this.firstNode = new Node(null, newNode);
            this.pushIncrement++;
            this.queueSemaphore.Release();
            return true;
        }
    }

    public T Pop()
    {
        this.queueSemaphore.WaitOne();
        lock (this.popLocker)
        {
            Node tempNext = this.firstNode.NextNode;
            T value = tempNext.Value;

            this.firstNode = tempNext;
            this.popIncrement++;
            return value;
        }
    }

    private sealed class Node
    {
        private readonly T value;

        private readonly Node nextNode;

        public Node()
        {
        }

        public Node(T value, Node nextNode)
        {
            this.value = value;
            this.nextNode = nextNode;
        }

        public T Value
        {
            get
            {
                return this.value;
            }
        }

        public Node NextNode
        {
            get
            {
                return this.nextNode;
            }
        }
    }
}