首页 > > 程序设计 > Java技术 >

并发编程（八）—— Java 并发队列 BlockingQueu…

2019-01-08 08:23:11来源：博客园阅读 ()

开篇先介绍下 BlockingQueue 这个接口的规则，后面再看其实现。

阻塞队列概要

阻塞队列与我们平常接触的普通队列(LinkedList或ArrayList等)的最大不同点，在于阻塞队列的阻塞添加和阻塞删除方法。

阻塞添加
所谓的阻塞添加是指当阻塞队列元素已满时，队列会阻塞加入元素的线程，直队列元素不满时才重新唤醒线程执行元素加入操作。

阻塞删除
阻塞删除是指在队列元素为空时，删除队列元素的线程将被阻塞，直到队列不为空再执行删除操作(一般都会返回被删除的元素)。

由于Java中的阻塞队列接口BlockingQueue继承自Queue接口，因此先来看看阻塞队列接口为我们提供的主要方法

 1 public interface BlockingQueue<E> extends Queue<E> {
 2 
 3     //将指定的元素插入到此队列的尾部（如果立即可行且不会超过该队列的容量）
 4     //在成功时返回 true，如果此队列已满，则抛IllegalStateException。 
 5     boolean add(E e); 
 6 
 7     //将指定的元素插入到此队列的尾部（如果立即可行且不会超过该队列的容量） 
 8     // 将指定的元素插入此队列的尾部，如果该队列已满， 
 9     //则在到达指定的等待时间之前等待可用的空间,该方法可中断 
10     boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException; 
11 
12     //将指定的元素插入此队列的尾部，如果该队列已满，则一直等到（阻塞）。 
13     void put(E e) throws InterruptedException; 
14 
15     //获取并移除此队列的头部，如果没有元素则等待（阻塞）， 
16     //直到有元素将唤醒等待线程执行该操作 
17     E take() throws InterruptedException; 
18 
19     //获取并移除此队列的头部，在指定的等待时间前一直等到获取元素， //超过时间方法将结束
20     E poll(long timeout, TimeUnit unit) throws InterruptedException; 
21 
22     //从此队列中移除指定元素的单个实例（如果存在）。 
23     boolean remove(Object o); 
24 }

这里我们把上述操作进行分类

插入方法：

　　add(E e) : 添加成功返回true，失败抛IllegalStateException异常
　　offer(E e) : 成功返回 true，如果此队列已满，则返回 false。
　　put(E e) :将元素插入此队列的尾部，如果该队列已满，则一直阻塞
删除方法：

　　remove(Object o) :移除指定元素,成功返回true，失败返回false
　　poll() : 获取并移除此队列的头元素，若队列为空，则返回 null
　　take()：获取并移除此队列头元素，若没有元素则一直阻塞。

阻塞队列的对元素的增删查操作主要就是上述的三类方法，通常情况下我们都是通过这3类方法操作阻塞队列，了解完阻塞队列的基本方法后，下面我们将分析阻塞队列中的两个实现类ArrayBlockingQueue和LinkedBlockingQueue的简单使用和实现原理，其中实现原理是这篇文章重点分析的内容。

ArrayBlockingQueue

在看源码之前，通过查询API发现对ArrayBlockingQueue特点的简单介绍：

1、一个由数组支持的有界队列，此队列按FIFO（先进先出）原则对元素进行排序。
2、新元素插入到队列的尾部，队列获取操作则是从队列头部开始获得元素
3、这是一个简单的“有界缓存区”，一旦创建，就不能在增加其容量
4、在向已满队列中添加元素会导致操作阻塞，从空队列中提取元素也将导致阻塞
5、此类支持对等待的生产者线程和使用者线程进行排序的可选公平策略。默认情况下，不保证是这种排序的。然而通过将公平性（fairness）设置为true，而构造的队列允许按照FIFO顺序访问线程。公平性通常会降低吞吐量，但也减少了可变性和避免了“不平衡性”。

简单的来说，ArrayBlockingQueue 是一个用数组实现的有界阻塞队列，其内部按先进先出的原则对元素进行排序，其中put方法和take方法为添加和删除的阻塞方法，下面我们通过ArrayBlockingQueue队列实现一个生产者消费者的案例，通过该案例简单了解其使用方式

使用示例

Consumer 消费者和 Producer 生产者，通过ArrayBlockingQueue 队列获取和添加元素，其中消费者调用了take()方法获取元素当队列没有元素就阻塞，生产者调用put()方法添加元素，当队列满时就阻塞，通过这种方式便实现生产者消费者模式。比直接使用等待唤醒机制或者Condition条件队列来得更加简单。

 1 package com.zejian.concurrencys.Queue;
 2 import java.util.concurrent.ArrayBlockingQueue;
 3 import java.util.concurrent.TimeUnit;
 4 
 5 /**
 6  * Created by chenhao on 2018/01/07
 7  */
 8 public class ArrayBlockingQueueDemo {
 9     private final static ArrayBlockingQueue<Apple> queue= new ArrayBlockingQueue<>(1);
10     public static void main(String[] args){
11         new Thread(new Producer(queue)).start();
12         new Thread(new Producer(queue)).start();
13         new Thread(new Consumer(queue)).start();
14         new Thread(new Consumer(queue)).start();
15     }
16 }
17 
18  class Apple {
19     public Apple(){
20     }
21  }
22 
23 /**
24  * 生产者线程
25  */
26 class Producer implements Runnable{
27     private final ArrayBlockingQueue<Apple> mAbq;
28     Producer(ArrayBlockingQueue<Apple> arrayBlockingQueue){
29         this.mAbq = arrayBlockingQueue;
30     }
31 
32     @Override
33     public void run() {
34         while (true) {
35             Produce();
36         }
37     }
38 
39     private void Produce(){
40         try {
41             Apple apple = new Apple();
42             mAbq.put(apple);
43             System.out.println("生产:"+apple);
44         } catch (InterruptedException e) {
45             e.printStackTrace();
46         }
47     }
48 }
49 
50 /**
51  * 消费者线程
52  */
53 class Consumer implements Runnable{
54 
55     private ArrayBlockingQueue<Apple> mAbq;
56     Consumer(ArrayBlockingQueue<Apple> arrayBlockingQueue){
57         this.mAbq = arrayBlockingQueue;
58     }
59 
60     @Override
61     public void run() {
62         while (true){
63             try {
64                 TimeUnit.MILLISECONDS.sleep(1000);
65                 comsume();
66             } catch (InterruptedException e) {
67                 e.printStackTrace();
68             }
69         }
70     }
71 
72     private void comsume() throws InterruptedException {
73         Apple apple = mAbq.take();
74         System.out.println("消费Apple="+apple);
75     }
76 }

输出：

1 生产:com.zejian.concurrencys.Queue.Apple@109967f
2 消费Apple=com.zejian.concurrencys.Queue.Apple@109967f
3 生产:com.zejian.concurrencys.Queue.Apple@269a77
4 生产:com.zejian.concurrencys.Queue.Apple@1ce746e
5 消费Apple=com.zejian.concurrencys.Queue.Apple@269a77
6 消费Apple=com.zejian.concurrencys.Queue.Apple@1ce746e
7 ........

源码剖析

ArrayBlockingQueue内部的阻塞队列是通过重入锁ReenterLock和Condition条件队列实现的，所以ArrayBlockingQueue中的元素存在公平访问与非公平访问的区别，对于公平访问队列，被阻塞的线程可以按照阻塞的先后顺序访问队列，即先阻塞的线程先访问队列。而非公平队列，当队列可用时，阻塞的线程将进入争夺访问资源的竞争中，也就是说谁先抢到谁就执行，没有固定的先后顺序。创建公平与非公平阻塞队列代码如下：

 1 //默认非公平阻塞队列
 2 ArrayBlockingQueue queue = new ArrayBlockingQueue(2);
 3 //公平阻塞队列
 4 ArrayBlockingQueue queue1 = new ArrayBlockingQueue(2,true);
 5 
 6 //构造方法源码
 7 public ArrayBlockingQueue(int capacity) {
 8      this(capacity, false);
 9  }
10 
11 public ArrayBlockingQueue(int capacity, boolean fair) {
12      if (capacity <= 0)
13          throw new IllegalArgumentException();
14      this.items = new Object[capacity];
15      lock = new ReentrantLock(fair);
16      notEmpty = lock.newCondition();
17      notFull =  lock.newCondition();
18  }

ArrayBlockingQueue的内部是通过一个可重入锁ReentrantLock和两个Condition条件对象来实现阻塞，这里先看看其内部成员变量

 1 public class ArrayBlockingQueue<E> extends AbstractQueue<E>
 2         implements BlockingQueue<E>, java.io.Serializable {
 3 
 4     /** 存储数据的数组 */
 5     final Object[] items;
 6 
 7     /**获取数据的索引，主要用于take，poll，peek，remove方法 */
 8     int takeIndex;
 9 
10     /**添加数据的索引，主要用于 put, offer, or add 方法*/
11     int putIndex;
12 
13     /** 队列元素的个数 */
14     int count;
15 
16 
17     /** 控制并非访问的锁 */
18     final ReentrantLock lock;
19 
20     /**notEmpty条件对象，用于通知take方法队列已有元素，可执行获取操作 */
21     private final Condition notEmpty;
22 
23     /**notFull条件对象，用于通知put方法队列未满，可执行添加操作 */
24     private final Condition notFull;
25 
26     /**
27        迭代器
28      */
29     transient Itrs itrs = null;
30 
31 }

从成员变量可看出，ArrayBlockingQueue内部确实是通过数组对象items来存储所有的数据，值得注意的是ArrayBlockingQueue通过一个ReentrantLock来同时控制添加线程与移除线程的并非访问，这点与LinkedBlockingQueue区别很大(稍后会分析)。而对于notEmpty条件对象则是用于存放等待或唤醒调用take方法的线程，告诉他们队列已有元素，可以执行获取操作。同理notFull条件对象是用于等待或唤醒调用put方法的线程，告诉它们，队列未满，可以执行添加元素的操作。takeIndex代表的是下一个方法(take，poll，peek，remove)被调用时获取数组元素的索引，putIndex则代表下一个方法（put, offer, or add）被调用时元素添加到数组中的索引。图示如下

添加

 1 //add方法实现，间接调用了offer(e)
 2 public boolean add(E e) {
 3         if (offer(e))
 4             return true;
 5         else
 6             throw new IllegalStateException("Queue full");
 7     }
 8 
 9 //offer方法
10 public boolean offer(E e) {
11      checkNotNull(e);//检查元素是否为null
12      final ReentrantLock lock = this.lock;
13      lock.lock();//加锁
14      try {
15          if (count == items.length)//判断队列是否满
16              return false;
17          else {
18              enqueue(e);//添加元素到队列
19              return true;
20          }
21      } finally {
22          lock.unlock();
23      }
24  }
25 
26 //入队操作
27 private void enqueue(E x) {
28     //获取当前数组
29     final Object[] items = this.items;
30     //通过putIndex索引对数组进行赋值
31     items[putIndex] = x;
32     //索引自增，如果已是最后一个位置，重新设置 putIndex = 0;
33     if (++putIndex == items.length)
34         putIndex = 0;
35     count++;//队列中元素数量加1
36     //唤醒调用take()方法的线程，执行元素获取操作。
37     notEmpty.signal();
38 }

这里的add方法和offer方法实现比较简单，其中需要注意的是enqueue(E x)方法，当putIndex索引大小等于数组长度时，需要将putIndex重新设置为0，因为后面讲到的取值也是从数组中第一个开始依次往后面取，取了之后会将原位置的值设置为null，方便循环put操作，这里要注意并不是每次都是取数组中的第一个值，takeIndex也会增加。因为做了添加操作，数组中肯定不会空，则 notEmpty条件会唤醒take()方法取值。

ok~,接着看put方法，它是一个阻塞添加的方法：

 1 //put方法，阻塞时可中断
 2  public void put(E e) throws InterruptedException {
 3      checkNotNull(e);
 4       final ReentrantLock lock = this.lock;
 5       lock.lockInterruptibly();//该方法可中断
 6       try {
 7           //当队列元素个数与数组长度相等时，无法添加元素
 8           while (count == items.length)
 9               //将当前调用线程挂起，添加到notFull条件队列中等待唤醒
10               notFull.await();
11           enqueue(e);//如果队列没有满直接添加。。
12       } finally {
13           lock.unlock();
14       }
15   }

put方法是一个阻塞的方法，如果队列元素已满，那么当前线程将会被notFull条件对象挂起加到等待队列中，直到队列有空档才会唤醒执行添加操作。但如果队列没有满，那么就直接调用enqueue(e)方法将元素加入到数组队列中。到此我们对三个添加方法即put，offer，add都分析完毕，其中offer，add在正常情况下都是无阻塞的添加，而put方法是阻塞添加。

（获取）删除

关于删除先看poll方法，该方法获取并移除此队列的头元素，若队列为空，则返回 null。

 1 public E poll() {
 2   final ReentrantLock lock = this.lock;
 3    lock.lock();
 4    try {
 5        //判断队列是否为null，不为null执行dequeue()方法，否则返回null
 6        return (count == 0) ? null : dequeue();
 7    } finally {
 8        lock.unlock();
 9    }
10 }
11  //删除队列头元素并返回
12  private E dequeue() {
13      //拿到当前数组的数据
14      final Object[] items = this.items;
15       @SuppressWarnings("unchecked")
16       //获取要删除的对象
17       E x = (E) items[takeIndex];
18       将数组中takeIndex索引位置设置为null
19       items[takeIndex] = null;
20       //takeIndex索引加1并判断是否与数组长度相等，
21       //如果相等说明已到尽头，恢复为0
22       if (++takeIndex == items.length)
23           takeIndex = 0;
24       count--;//队列个数减1
25       if (itrs != null)
26           itrs.elementDequeued();//同时更新迭代器中的元素数据
27       //删除了元素说明队列有空位，唤醒notFull条件对象添加线程，执行添加操作
28       notFull.signal();
29       return x;
30  }

接着看take()方法，是一个阻塞方法，获取队列头元素并删除。

 1 //从队列头部删除，队列没有元素就阻塞，可中断
 2  public E take() throws InterruptedException {
 3     final ReentrantLock lock = this.lock;
 4       lock.lockInterruptibly();//中断
 5       try {
 6           //如果队列没有元素
 7           while (count == 0)
 8               //执行阻塞操作
 9               notEmpty.await();
10           return dequeue();//如果队列有元素执行删除操作
11       } finally {
12           lock.unlock();
13       }
14  }

take和poll的区别是，队列为空时，poll返回null，take则被挂起阻塞，直到有元素添加进来，take线程被唤醒，然后获取第一个元素并删除。

peek方法非常简单，直接返回当前队列的头元素但不删除任何元素。

 1 public E peek() {
 2       final ReentrantLock lock = this.lock;
 3       lock.lock();
 4       try {
 5        //直接返回当前队列的头元素，但不删除
 6           return itemAt(takeIndex); // null when queue is empty
 7       } finally {
 8           lock.unlock();
 9       }
10   }
11 
12 final E itemAt(int i) {
13       return (E) items[i];
14   }