【java】Java多线程总结之线程安全队列Queue【转载】

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

原文地址：https://www.cnblogs.com/java-jun-world2099/articles/10165949.html

========================================================================

在Java多线程应用中，队列的使用率很高，多数生产消费模型的首选数据结构就是队列。Java提供的线程安全的Queue可以分为阻塞队列和非阻塞队列，其中阻塞队列的典型例子是BlockingQueue，非阻塞队列的典型例子是

ConcurrentLinkedQueue，在实际应用中要根据实际需要选用阻塞队列或者非阻塞队列。

注：什么叫线程安全？这个首先要明确。线程安全的类，指的是类内共享的全局变量的访问必须保证是不受多线程形式影响的。如果由于多线程的访问（比如修改、遍历、查看）而使这些变量结构被破坏或者针对这些变量操作的原子性被破坏，则这个类就不是线程安全的。

在并发的队列上jdk提供了两套实现，一个是以ConcurrentLinkedQueue为代表的高性能队列，一个是以BlockingQueue接口为代表的阻塞队列，无论在那种都继承自Queue。
今天就聊聊这两种Queue：

BlockingQueue 阻塞算法
ConcurrentLinkedQueue，非阻塞算法

一、首先来看看BlockingQueue：

Queue是什么就不需要多说了吧，一句话：队列是先进先出。相对的，栈是后进先出。如果不熟悉的话先找本基础的数据结构的书看看吧。

BlockingQueue，顾名思义，“阻塞队列”：可以提供阻塞功能的队列。
首先，看看BlockingQueue提供的常用方法：

从上表可以很明显看出每个方法的作用，这个不用多说。我想说的是：

add(e) remove() element() 方法不会阻塞线程。当不满足约束条件时，会抛出IllegalStateException 异常。例如：当队列被元素填满后，再调用add(e)，则会抛出异常。
offer(e) poll() peek() 方法即不会阻塞线程，也不会抛出异常。例如：当队列被元素填满后，再调用offer(e)，则不会插入元素，函数返回false。
要想要实现阻塞功能，需要调用put(e) take() 方法。当不满足约束条件时，会阻塞线程。

BlockingQueue 阻塞算法

BlockingQueue作为线程容器，可以为线程同步提供有力的保障。

BlockingQueue定义的常用方法：

抛出异常特殊值阻塞超时

插入 add(e) offer(e) put(e) offer(e, time, unit)
移除 remove() poll() take() poll(time, unit)
检查 element() peek() 不可用不可用

1、ArrayBlockingQueue

基于数组的阻塞队列实现，在ArrayBlockingQueue内部，维护了一个定长数组，以便缓存队列中的数据对象，这是一个常用的阻塞队列，除了一个定长数组外，ArrayBlockingQueue内部还保存着两个整形变量，分别标识着队列的头部和尾部在数组中的位置。
ArrayBlockingQueue在生产者放入数据和消费者获取数据，都是共用同一个锁对象，由此也意味着两者无法真正并行运行，这点尤其不同于LinkedBlockingQueue；按照实现原理来分析，ArrayBlockingQueue完全可以采用分离锁，从而实现生产者和消费者操作的完全并行运行。Doug Lea之所以没这样去做，也许是因为ArrayBlockingQueue的数据写入和获取操作已经足够轻巧，以至于引入独立的锁机制，除了给代码带来额外的复杂性外，其在性能上完全占不到任何便宜。 ArrayBlockingQueue和LinkedBlockingQueue间还有一个明显的不同之处在于，前者在插入或删除元素时不会产生或销毁任何额外的对象实例，而后者则会生成一个额外的Node对象。这在长时间内需要高效并发地处理大批量数据的系统中，其对于GC的影响还是存在一定的区别。而在创建ArrayBlockingQueue时，我们还可以控制对象的内部锁是否采用公平锁，默认采用非公平锁。

2、LinkedBlockingQueue

基于链表的阻塞队列，同ArrayListBlockingQueue类似，其内部也维持着一个数据缓冲队列（该队列由一个链表构成），当生产者往队列中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者立即返回；只有当队列缓冲区达到最大值缓存容量时（LinkedBlockingQueue可以通过构造函数指定该值），才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生产者线程会被唤醒，反之对于消费者这端的处理也基于同样的原理。而LinkedBlockingQueue之所以能够高效的处理并发数据，还因为其对于生产者端和消费者端分别采用了独立的锁来控制数据同步，这也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。
作为开发者，我们需要注意的是，如果构造一个LinkedBlockingQueue对象，而没有指定其容量大小，LinkedBlockingQueue会默认一个类似无限大小的容量（Integer.MAX_VALUE），这样的话，如果生产者的速度一旦大于消费者的速度，也许还没有等到队列满阻塞产生，系统内存就有可能已被消耗殆尽了。

阻塞队列：线程安全

按 FIFO（先进先出）排序元素。队列的头部是在队列中时间最长的元素。队列的尾部是在队列中时间最短的元素。新元素插入到队列的尾部，并且队列检索操作会获得位于队列头部的元素。链接队列的吞吐量通常要高于基于数组的队列，但是在大多数并发应用程序中，其可预知的性能要低。

注意：

1、必须要使用take()方法在获取的时候达成阻塞结果
2、使用poll()方法将产生非阻塞效果

3、LinkedBlockingQueue实例

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;public class BlockingDeque {
//阻塞队列，FIFO
private static LinkedBlockingQueue<Integer> concurrentLinkedQueue = new LinkedBlockingQueue<Integer>(); public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(2); executorService.submit(new Producer("producer1"));
executorService.submit(new Producer("producer2"));
executorService.submit(new Producer("producer3"));
executorService.submit(new Consumer("consumer1"));
executorService.submit(new Consumer("consumer2"));
executorService.submit(new Consumer("consumer3")); } static class Producer implements Runnable {
private String name; public Producer(String name) {
this.name = name;
} public void run() {
for (int i = 1; i < 10; ++i) {
System.out.println(name+ " 生产： " + i);
//concurrentLinkedQueue.add(i);
try {
concurrentLinkedQueue.put(i);
Thread.sleep(200); //模拟慢速的生产，产生阻塞的效果
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
} }
}
} static class Consumer implements Runnable {
private String name; public Consumer(String name) {
this.name = name;
}
public void run() {
for (int i = 1; i < 10; ++i) {
try {
//必须要使用take()方法在获取的时候阻塞
System.out.println(name+"消费： " + concurrentLinkedQueue.take());
//使用poll()方法将产生非阻塞效果
//System.out.println(name+"消费： " + concurrentLinkedQueue.poll()); //还有一个超时的用法，队列空时，指定阻塞时间后返回，不会一直阻塞
//但有一个疑问，既然可以不阻塞，为啥还叫阻塞队列？
//System.out.println(name+" Consumer " + concurrentLinkedQueue.poll(300, TimeUnit.MILLISECONDS));
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
} }
}
}
}

4、PriorityBlockingQueue

基于优先级的阻塞队列（优先级的判断通过构造函数传入的Compator对象来决定，也就是说传入队列的对象必须实现Comparable接口），在实现PriorityBlockingQueue时，内部控制线程同步的锁采用的是公平锁，他也是一个无界的队列。

5、PriorityBlockingQueue 实例

Task.java

UsePriorityBlockingQueue.java

打印结果：

6、DelayQueue

带有延迟时间的Queue，其中的元素只有当其指定的延迟时间到了，才能够从队列中获取到该元素。DelayQueue中的元素必须实现Delayed接口，DelayQueue是一个没有大小限制的队列，应用场景很多，比如对缓存超时的数据进行移除、任务超时处理、空闲连接的关闭等等。

7、DelayQueue实例

Wangmin.java

WangBa.java

打印结果：

8、LinkedBlockingDeque

LinkedBlockingDeque是一个线程安全的双端队列实现，由链表结构组成的双向阻塞队列，即可以从队列的两端插入和移除元素。双向队列因为多了一个操作队列的入口，在多线程同时入队时，也就减少了一半的竞争。可以说他是最为复杂的一种队列，在内部实现维护了前端和后端节点，但是其没有实现读写分离，因此同一时间只能有一个线程对其讲行操作。在高并发中性能要远低于其它BlockingQueue。更要低于ConcurrentLinkedQueue，jdk早期有一个非线程安全的Deque就是ArryDeque了， java6里添加了LinkBlockingDeque来弥补多线程场景下线程安全的问题。

相比于其他阻塞队列，LinkedBlockingDeque多了addFirst、addLast、peekFirst、peekLast等方法，以first结尾的方法，表示插入、获取获移除双端队列的第一个元素。以last结尾的方法，表示插入、获取获移除双端队列的最后一个元素。

此外，LinkedBlockingDeque还是可选容量的,防止过度膨胀，默认等于Integer.MAX_VALUE。

主要方法：

akeFirst()和takeLast()：分别返回类表中第一个和最后一个元素，返回的元素会从类表中移除。如果列表为空，调用的方法的线程将会被阻塞直达列表中有可用元素。

getFirst()和getLast():分别返回类表中第一个和最后一个元素，返回的元素不会从列表中移除。如果列表为空，则抛出NoSuckElementException异常。

peek()、peekFirst()和peekLast():分别返回列表中第一个元素和最后一个元素，返回元素不会被移除。如果列表为空返回null。

poll()、pollFirst()和pollLast():分别返回类表中第一个和最后一个元素，返回的元素会从列表中移除。如果列表为空，返回Null。

public class UseDeque {
public static void main(String[] args) {
LinkedBlockingDeque<String> dq = new LinkedBlockingDeque<String>(10);
dq.addFirst("a");
dq.addFirst("b");
dq.addFirst("c");
dq.addFirst("d");
dq.addFirst("e");
dq.addLast("f");
dq.addLast("g");
dq.addLast("h");
dq.addLast("i");
dq.addLast("j");
//dq.offerFirst("k");
System.out.println("查看头元素：" + dq.peekFirst());
System.out.println("获取尾元素：" + dq.pollLast());
Object [] objs = dq.toArray();
for (int i = 0; i < objs.length; i++) {
System.out.print(objs[i] + " -- ");
}
}
}

打印结果：

9、LinkedBlockingDeque方法列表

10、LinkedBlockingQeque和LinkedBlockingDeque源码解读

1）LinkedBlockingQeque

先看它的结构基本字段：

和LinkedBlockingDeque的区别之一就是，LinkedBlockingQueue采用了两把锁来对队列进行操作，也就是队尾添加的时候，

队头仍然可以删除等操作。接下来看典型的操作。

put操作

主要的思想还是比较容易理解的，现在看看enqueue 方法：

private void enqueue(Node<E> node) { //入对操作。
last = last.next = node; //队尾进
}

再看看signalNotEmpty方法：

private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock(); //加锁
try {
notEmpty.signal(); //用于signal，notEmpty
} finally {
takeLock.unlock();
}
}

take操作

take操作，就是从队列里面弹出一个元素，下面看它的详细代码：

public E take() throws InterruptedException {
E x;
int c = -1; //设定一个记录变量
final AtomicInteger count = this.count; //获得count
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly(); //加锁
try {
while (count.get() == 0) { //如果没有元素，那么就阻塞性等待
notEmpty.await();
}
x = dequeue(); //一定可以拿到。
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal(); //报告还有元素，唤醒队列
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull(); //解锁
return x;
}

接下来看dequeue方法：

private E dequeue() {
Node<E> h = head;
Node<E> first = h.next;
h.next = h; // help GC 指向自己，帮助gc回收
head = first;
E x = first.item; //从队头出。
first.item = null; //将head.item设为null。
return x;
}

对于LinkedBlockingQueue来说，有两个ReentrantLock分别控制队头和队尾，这样就可以使得添加操作分开来做，一般的操作是获取一把锁就可以，但有些操作例如remove操作，则需要同时获取两把锁：

public boolean remove(Object o) {
if (o == null) return false;
fullyLock(); //获取锁
try {
for (Node<E> trail = head, p = trail.next;
p != null;
trail = p, p = p.next) { //依次循环遍历
if (o.equals(p.item)) { //找到了
unlink(p, trail); //解除链接
return true;
}
}
return false; //没找到，或者解除失败
} finally {
fullyUnlock();
}
}

当然，除了上述的remove方法外，在Iterator的next方法，remove方法以及LBQSpliterator分割迭代器中也是需要加全锁进行操作的。

2）LinkedBlockingDeque

LinkedBlockingDeque类有三个构造方法：

public LinkedBlockingDeque()
public LinkedBlockingDeque(int capacity)
public LinkedBlockingDeque(Collection<? extends E> c)

LinkedBlockingDeque类中的数据都被封装成了Node对象：

static final class Node<E> {
E item;
Node<E> prev;
Node<E> next; Node(E x) {
item = x;
}
}

LinkedBlockingDeque类中的重要字段如下：

// 队列双向链表首节点
transient Node<E> first;
// 队列双向链表尾节点
transient Node<E> last;
// 双向链表元素个数
private transient int count;
// 双向链表最大容量
private final int capacity;
// 全局独占锁
final ReentrantLock lock = new ReentrantLock();
// 非空Condition对象
private final Condition notEmpty = lock.newCondition();
// 非满Condition对象
private final Condition notFull = lock.newCondition();

LinkedBlockingDeque类的底层实现和LinkedBlockingQueue类很相似，都有一个全局独占锁，和两个Condition对象，用来阻塞和唤醒线程。

LinkedBlockingDeque类对元素的操作方法比较多，我们下面以putFirst、putLast、pollFirst、pollLast方法来对元素的入队、出队操作进行分析。

入队

putFirst(E e)方法是将指定的元素插入双端队列的开头，源码如下：

public void putFirst(E e) throws InterruptedException {
// 若插入元素为null，则直接抛出NullPointerException异常
if (e == null) throw new NullPointerException();
// 将插入节点包装为Node节点
Node<E> node = new Node<E>(e);
// 获取全局独占锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
while (!linkFirst(node))
notFull.await();
} finally {
// 释放全局独占锁
lock.unlock();
}
}

入队操作是通过linkFirst(E e)方法来完成的，如下所示：

private boolean linkFirst(Node<E> node) {
// assert lock.isHeldByCurrentThread();
// 元素个数超出容量。直接返回false
if (count >= capacity)
return false;
// 获取双向链表的首节点
Node<E> f = first;
// 将node设置为首节点
node.next = f;
first = node;
// 若last为null，设置尾节点为node节点
if (last == null)
last = node;
else
// 更新原首节点的前驱节点
f.prev = node;
++count;
// 唤醒阻塞在notEmpty上的线程
notEmpty.signal();
return true;
}

若入队成功，则linkFirst(E e)方法返回true，否则，返回false。若该方法返回false，则当前线程会阻塞在notFull条件上。

putLast(E e)方法是将指定的元素插入到双端队列的末尾，源码如下：

public void putLast(E e) throws InterruptedException {
// 若插入元素为null，则直接抛出NullPointerException异常
if (e == null) throw new NullPointerException();
// 将插入节点包装为Node节点
Node<E> node = new Node<E>(e);
// 获取全局独占锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
while (!linkLast(node))
notFull.await();
} finally {
// 释放全局独占锁
lock.unlock();
}
}

该方法和putFirst(E e)方法几乎一样，不同点在于，putLast(E e)方法通过调用linkLast(E e)方法来插入节点：

private boolean linkLast(Node<E> node) {
// assert lock.isHeldByCurrentThread();
// 元素个数超出容量。直接返回false
if (count >= capacity)
return false;
// 获取双向链表的尾节点
Node<E> l = last;
// 将node设置为尾节点
node.prev = l;
last = node;
// 若first为null，设置首节点为node节点
if (first == null)
first = node;
else
// 更新原尾节点的后继节点
l.next = node;
++count;
// 唤醒阻塞在notEmpty上的线程
notEmpty.signal();
return true;
}

若入队成功，则linkLast(E e)方法返回true，否则，返回false。若该方法返回false，则当前线程会阻塞在notFull条件上。

出队

pollFirst()方法是获取并移除此双端队列的首节点，若不存在，则返回null，源码如下：

public E pollFirst() {
// 获取全局独占锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
return unlinkFirst();
} finally {
// 释放全局独占锁
lock.unlock();
}
}

移除首节点的操作是通过unlinkFirst()方法来完成的：

private E unlinkFirst() {
// assert lock.isHeldByCurrentThread();
// 获取首节点
Node<E> f = first;
// 首节点为null，则返回null
if (f == null)
return null;
// 获取首节点的后继节点
Node<E> n = f.next;
// 移除first，将首节点更新为n
E item = f.item;
f.item = null;
f.next = f; // help GC
first = n;
// 移除首节点后，为空队列
if (n == null)
last = null;
else
// 将新的首节点的前驱节点设置为null
n.prev = null;
--count;
// 唤醒阻塞在notFull上的线程
notFull.signal();
return item;
}

pollLast()方法是获取并移除此双端队列的尾节点，若不存在，则返回null，源码如下：

public E pollLast() {
// 获取全局独占锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
return unlinkLast();
} finally {
// 释放全局独占锁
lock.unlock();
}
}

移除尾节点的操作是通过unlinkLast()方法来完成的：

private E unlinkLast() {
// assert lock.isHeldByCurrentThread();
// 获取尾节点
Node<E> l = last;
// 尾节点为null，则返回null
if (l == null)
return null;
// 获取尾节点的前驱节点
Node<E> p = l.prev;
// 移除尾节点，将尾节点更新为p
E item = l.item;
l.item = null;
l.prev = l; // help GC
last = p;
// 移除尾节点后，为空队列
if (p == null)
first = null;
else
// 将新的尾节点的后继节点设置为null
p.next = null;
--count;
// 唤醒阻塞在notFull上的线程
notFull.signal();
return item;
}

其实LinkedBlockingDeque类的入队、出队操作都是通过linkFirst、linkLast、unlinkFirst、unlinkLast这几个方法来实现的，源码读起来也比较简单。

二、ConcurrentLinkedQueue 非阻塞算法

1、非阻塞队列

基于链接节点的、无界的、线程安全。此队列按照 FIFO（先进先出）原则对元素进行排序。队列的头部是队列中时间最长的元素。队列的尾部是队列中时间最短的元素。新的元素插入到队列的尾部，队列检索操作从队列头部获得元素。当许多线程共享访问一个公共 collection 时，ConcurrentLinkedQueue 是一个恰当的选择。此队列不允许 null 元素。

ConcurrentLinkedQueue是一个适用于高并发场景下的队列，通过无锁的方式，实现了高并发状态下的高性能，通常ConcurrentLinkedQueue性能好于BlockingQueue。

ConcurrentLinkedQueue重要方法：

add()和offer()都是加入元素的方法（在ConcurrentLinkedQueue中，这两个方法投有任何区别）

poll()和peek()都是取头元素节点，区别在于前者会删除元素，后者不会，相当于查看。

public class UseQueue_ConcurrentLinkedQueue { public static void main(String[] args) throws Exception { //高性能无阻塞无界队列：ConcurrentLinkedQueue ConcurrentLinkedQueue<String> q = new ConcurrentLinkedQueue<String>();
q.offer("a");
q.offer("b");
q.offer("c");
q.offer("d");
q.add("e"); System.out.println("从头部取出元素，并从队列里删除 >> "+q.poll()); //a 从头部取出元素，并从队列里删除
System.out.println("删除后的长度 >> "+q.size()); //4
System.out.println("取出头部元素 >> "+q.peek()); //b
System.out.println("长度 >> "+q.size()); //4
}
}

打印结果：

2、实例

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;public class NoBlockQueue {
private static ConcurrentLinkedQueue<Integer> concurrentLinkedQueue = new ConcurrentLinkedQueue<Integer>(); public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(2); executorService.submit(new Producer("producer1"));
executorService.submit(new Producer("producer2"));
executorService.submit(new Producer("producer3"));
executorService.submit(new Consumer("consumer1"));
executorService.submit(new Consumer("consumer2"));
executorService.submit(new Consumer("consumer3")); } static class Producer implements Runnable {
private String name; public Producer(String name) {
this.name = name;
} public void run() {
for (int i = 1; i < 10; ++i) {
System.out.println(name+ " start producer " + i);
concurrentLinkedQueue.add(i);
try {
Thread.sleep(20);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
//System.out.println(name+"end producer " + i);
}
}
} static class Consumer implements Runnable {
private String name; public Consumer(String name) {
this.name = name;
}
public void run() {
for (int i = 1; i < 10; ++i) {
try { System.out.println(name+" Consumer " + concurrentLinkedQueue.poll()); } catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
// System.out.println();
// System.out.println(name+" end Consumer " + i);
}
}
}
}