java - BlockingQueue 在 Java 中是完全线程安全的吗

Question

我知道文档说该对象是线程安全的，但这是否意味着从所有方法对其进行的所有访问都是线程安全的？因此，如果我同时从多个线程调用 put() 并在同一实例上调用 take() ，不会发生任何不好的事情吗？

我问是因为这个答案让我第二次猜测: https://stackoverflow.com/a/22006181/4164238

Answer 1

答案是肯定的，它们是线程安全的。但我们不要把它留在那里......

首先做一点小事，BlockingQueue 是一个接口(interface)，任何不是线程安全的实现都将违反记录在案的契约(Contract)。您包含的链接是指 LinkedBlockingQueue，它有一些聪明之处。

link that you included做了一个有趣的观察，是的，LinkedBlockingQueue 中有两个锁。然而，它无法理解“简单”实现会遇到的边缘情况实际上正在被处理，这就是为什么 take 和 put 方法比人们最初预期的要复杂的原因。

LinkedBlockingQueue 进行了优化，以避免在读取和写入时使用相同的锁，这减少了争用，但是对于正确的行为，它依赖于队列不为空。当队列中有元素时，推送点和弹出点不在同一内存区域，可以避免争用。但是，当队列为空时，就无法避免争用，因此需要额外的代码来处理这种常见的“边缘”情况。这是代码复杂性和性能/可扩展性之间的常见权衡。

接下来的问题是，LinkedBlockingQueue 如何知道队列何时为空/非空，从而处理线程呢？答案是它使用 AtomicInteger 和 Condition 作为两个额外的并发数据结构。 AtomicInteger 用于检查队列的长度是否为零，Condition 用于等待队列可能处于所需状态时通知等待线程的信号。这种额外的协调确实有开销，但是在测量中表明，当增加并发线程的数量时，这种技术的开销低于使用单个锁引入的争用。

下面我复制了 LinkedBlockingQueue 中的代码，并添加了解释它们如何工作的注释。在高层次上，take() 首先锁定所有其他对 take() 的调用，然后根据需要向 put() 发出信号。 put() 以类似的方式工作，首先它阻止所有其他对 put() 的调用，然后在必要时发出 take() 信号。

来自 put() 方法:

    // putLock coordinates the calls to put() only; further coordination
    // between put() and take() follows below
    putLock.lockInterruptibly();
    try {
        // block while the queue is full; count is shared between put() and take()
        // and is safely visible between cores but prone to change between calls
        // a while loop is used because state can change between signals, which is
        // why signals get rechecked and resent.. read on to see more of that 
        while (count.get() == capacity) { 
                notFull.await();
        }

        // we know that the queue is not full so add
        enqueue(e);
        c = count.getAndIncrement();

        // if the queue is not full, send a signal to wake up 
        // any thread that is possibly waiting for the queue to be a little
        // emptier -- note that this is logically part of 'take()' but it
        // has to be here because take() blocks itself
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        signalNotEmpty();

来自take()

    takeLock.lockInterruptibly();
    try {
            // wait for the queue to stop being empty
            while (count.get() == 0) {
                notEmpty.await();
            }

        // remove element
        x = dequeue();

        // decrement shared count
        c = count.getAndDecrement();

        // send signal that the queue is not empty
        // note that this is logically part of put(), but
        // for thread coordination reasons is here
        if (c > 1)
            notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    if (c == capacity)
        signalNotFull();