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

Question

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

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

Answer 1

我认为@Chris K错过了一些要点。“当队列中有元素时，推送点和弹出点不在同一内存区域，可以避免争用。”，请注意，当队列有一个元素时，head。next和tail指向同一个节点，put（）和take（）都可以获得锁并执行

我认为空和满条件可以通过同步put（）和take（）解决。但是，对于一个元素，lb队列有一个空的虚拟头节点，这可能与线程安全有关

Answer 2

我在Leetcode上尝试了这个实现 导入java。util。同时发生的阻塞队列； 导入java。util。同时发生的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();
        }
    }
}

使用n=3，大多数时候我会得到foobarfoobarfoorbar的正确响应，但有时我会得到barbarfoobar，这非常令人惊讶。 我决定使用ReentrantLock和Condition，@chris-k你能透露更多信息吗

Answer 3

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

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

Answer 4

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

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

{a1}是一个有趣的观察，是的，{}中有两个锁。然而，它无法理解，一个“简单”实现可能会遇到麻烦的边缘案例实际上正在处理中，这就是take-and-put方法比人们最初预期的更复杂的原因

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

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

下面，我从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();

Answer 5

这个答案有点奇怪——首先，BlockingQueue是一个接口，所以它没有任何锁。诸如ArrayBlockingQueue之类的实现对add（）和take（）使用相同的锁，因此可以。通常，如果任何实现都不是线程安全的，那么它就是一个有缺陷的实现