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

Is BlockingQueue completely thread safe in Java

提问by Josh

I know that the documentation says that the object is thread safe but does that mean that all access to it from all methods are thread safe? So if I call put()on it from many threads at once and take()on it at the same instance, will nothing bad happen?

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

I ask because this answer is making me second guess: https://stackoverflow.com/a/22006181/4164238

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

回答by Chris K

The quick answer is yes, they are thread safe. But lets not leave it there ...

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

Firstly a little house keeping, BlockingQueueis an interface, and any implementation that is not thread safe will be breaking the documented contract. The link that you included was referring to LinkedBlockingQueue, which has some cleverness to it.

首先是一个小家务，BlockingQueue是一个接口，任何不是线程安全的实现都将破坏记录在案的契约。您包含的链接指的是LinkedBlockingQueue，它有一些巧妙之处。

The link that you includedmakes an interesting observation, yes there are two locks within LinkedBlockingQueue. However it fails to understand that the edge case that a 'simple' implementation would have fallen foul of was in-fact being handled, which is why the take and put methods are more complicated than one would at first expect.

您包含的链接进行了有趣的观察，是的，LinkedBlockingQueue. 然而，它无法理解实际上正在处理“简单”实现会犯规的边缘情况，这就是为什么 take 和 put 方法比人们最初预期的更复杂。

LinkedBlockingQueueis optimized to avoid using the same lock on both reading and writing, this reduces contention however for correct behavior it relies on the queue not being empty. When the queue has elements within it, then the push and the pop points are not at the same region of memory and contention can be avoided. However when the queue is empty then the contention cannot be avoided, and so extra code is required to handle this common 'edge' case. This is a common trade off between code complexity and performance/scalability.

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

The question then follows, how does LinkedBlockingQueueknow when the queue is empty/not empty and thus handle the threading then? The answer is that it uses an AtomicIntegerand a Conditionas two extra concurrent data structures. The AtomicIntegeris used to check whether the length of the queue is zero and the Condition is used to wait for a signal to notify a waiting thread when the queue is probably in the desired state. This extra coordination does have an overhead, however in measurements it has been shown that when ramping up the number of concurrent threads that the overheads of this technique are lower than the contention that is introduced by using a single lock.

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

Below I have copied the code from LinkedBlockingQueueand added comments explaining how they work. At a high level, take()first locks out all other calls to take()and then signals put()as necessary. put()works in a similar way, first it blocks out all other calls to put()and then signals take()if necessary.

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

From the put()method:

从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();

From take()

从 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();

回答by apadana

Yes, all implementations of BlockingQueue are thread safe for put and take and all actions.

是的，BlockingQueue 的所有实现对于 put 和 take 以及所有操作都是线程安全的。

The link just goes halfway...and is not covering the full details. It is thread safe.

该链接只是进行了一半……并没有涵盖全部细节。它是线程安全的。

回答by asap diablo

I think @Chris K has missed some points. "When the queue has elements within it, then the push and the pop points are not at the same region of memory and contention can be avoided. ", notice that when the queue has one element, head.next and tail points to the same node and put() and take() can both get locks and execute.

我认为@Chris K 错过了一些要点。“当队列中有元素时，push点和pop点不在同一个内存区域，可以避免争用。”，注意当队列有一个元素时，head.next和tail指向同一个node 和 put() 和 take() 都可以获取锁并执行。

I think empty and full condition can be solved by synchronized put() and take(). However when it comes to one element, the lb queue has a null dummy head node, which may has something to do with the thread safety.

我认为可以通过同步 put() 和 take() 来解决空和满的情况。但是当涉及到一个元素时，lb 队列有一个空的虚拟头节点，这可能与线程安全有关。

回答by Akinniranye James

I tried this implementation on Leetcodeimport java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingDeque;

我在Leetcodeimport java.util.concurrent.BlockingQueue;上尝试了这个实现；导入 java.util.concurrent.LinkedBlockingDeque;

class FooBar {

    private final BlockingQueue<Object> line =  new LinkedBlockingDeque<>(1);
    private static final Object PRESENT = new Object();
    private int n;

    public FooBar(int n) {
        this.n = n;
    }

    public void foo(Runnable printFoo) throws InterruptedException {

        for (int i = 0; i < n; i++) {
            line.put(PRESENT);
            // printFoo.run() outputs "foo". Do not change or remove this line.
            printFoo.run();
        }
    }

    public void bar(Runnable printBar) throws InterruptedException {

        for (int i = 0; i < n; i++) {
            line.take();
            // printBar.run() outputs "bar". Do not change or remove this line.
            printBar.run();
        }
    }
}

With n = 3, mosttimes I get a correct response of foobarfoobarfoorbarbut sometimes I get barbarfoofoofoobarwhich is quite surprising. I resolved to use using ReentrantLock and Condition, @chris-k can you shed more light

随着N = 3，mosttimes我得到一个正确的响应foob​​arfoobarfoorbar但有时我得到barbarfoofoofoobar这是相当惊人。我决定使用 ReentrantLock 和 Condition，@chris-k 你能解释一下吗

回答by BarrySW19

That answer is a little strange - for a start, BlockingQueue is an interface so it doesn't have any locks. Implementations such as ArrayBlockingQueue use the same lock for add() and take() so would be fine. Generally, if any implementation is not thread safe then it is a buggy implementation.

这个答案有点奇怪——首先，BlockingQueue 是一个接口，所以它没有任何锁。ArrayBlockingQueue 等实现对 add() 和 take() 使用相同的锁，所以没问题。通常，如果任何实现不是线程安全的，那么它就是一个有缺陷的实现。