在Java中，BlockingQueue是否完全是线程安全的

本文介绍了在Java中，BlockingQueue是否完全是线程安全的的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

我知道文档说该对象是线程安全的，但这是否意味着从所有方法对它的所有访问都是线程安全的？所以，如果我一次从多个线程调用put（）并在同一个实例上调用（），那么会发生什么不好吗？

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?

我问因为这个答案让我猜第二个猜测： stackoverflow/a/22006181/4164238

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

推荐答案

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

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

首先是一个小房子， BlockingQueue 是一个界面，任何非线程安全的实现将破坏记录的合同。你包含的链接指的是 LinkedBlockingQueue ，它有一些巧妙。

Firstly a little house keeping, BlockingQueue is 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.

你包含的链接是一个有趣的观察，是的，有 LinkedBlockingQueue 中的两个锁。然而，它无法理解简单实现可能违反的边缘情况实际上正在被处理，这就是为什么take和put方法比最初期望的更复杂。

The link that you included makes 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 经过优化，以避免在读取和写入时使用相同的锁定，这样可以减少争用，但是对于正确的行为，它依赖于队列不是空的。当队列中包含元素时，推送和弹出点不在同一内存区域，并且可以避免争用。但是当队列为空时，无法避免争用，因此需要额外的代码来处理这种常见的边缘情况。这是代码复杂性和性能/可伸缩性之间的常见折衷。

LinkedBlockingQueue is 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 知道什么时候队列是空的/非空的，从而处理线程呢？答案是它使用 AtomicInteger 和 Condition 作为两个额外的并发数据结构。 AtomicInteger 用于检查队列长度是否为零，条件用于等待信号在队列可能需要时通知等待线程州。这种额外的协调确实有开销，但是在测量中已经表明，当增加并发线程的数量时，这种技术的开销低于使用单个锁引入的争用。

The question then follows, how does LinkedBlockingQueue know when the queue is empty/not empty and thus handle the threading then? The answer is that it uses an AtomicInteger and a Condition as two extra concurrent data structures. The AtomicInteger is 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 的代码，并添加了解释它们如何工作的注释。在较高级别， take（）首先锁定对 take（）的所有其他调用，然后发出信号 put（）根据需要。 put（）以类似的方式工作，首先阻止对 put（）的所有其他调用，然后发出信号 take（）如有必要。

Below I have copied the code from LinkedBlockingQueue and 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.

来自 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();