Java中的BlockingQueue队列

Posted 2020-11-03 奔跑在梦想的道路上

　　BlockingQueue位于JDK5新增的concurrent包中，它很好地解决了多线程中，如何高效安全地“传输”数据的问题。通过这些高效并且线程安全的队列类，为我们快速搭建高质量的多线程程序带来极大的便利。

　　阻塞队列，顾名思义，它首先它是一个队列，在数据结构中，队列是一种线性表。

　　我们通过一个共享的队列，可以使得数据由队列的一端输入，从另外一端输出。常用的队列主要有以下两种：
　　先进先出（FIFO）：先插入的队列的元素也最先出队列，类似于排队的功能。从某种程度上来说这种队列也体现了一种公平性。
　　后进先出（LIFO）：后插入队列的元素最先出队列，这种队列优先处理最近发生的事件，相当于栈。

　　在多线程环境中，通过队列可以很容易地实现数据共享，比如经典的“生产者”和“消费者”模型中，通过队列可以很便利地实现两者之间的数据共享。

　　假设我们有若干生产者线程，另外又有若干个消费者线程。如果生产者线程需要把准备好的数据共享给消费者线程，利用队列的方式来传递数据，就可以很方便地解决他们之间的数据共享问题。但是，如果生产者和消费者在某个时间段内，万一发生数据处理速度不匹配的情况呢？理想情况下，如果生产者产出数据的速度大于消费者消费的速度，并且当生产出来的数据累积到一定程度的时候，那么生产者必须暂停等待一下（阻塞生产者线程），以便等待消费者线程把累积的数据处理完毕，反之亦然。

　　在JDK5的concurrent包发布以前，在多线程环境下，程序员只能靠自己去人为地控制这些实现细节，还要兼顾效率和线程安全，这会给我们的程序带来不小的复杂度。于是，强大的concurrent包横空出世了，并且给我们带来了强大的BlockingQueue。（注：在多线程领域，所谓阻塞，在某些情况下会挂起线程（即阻塞），一旦条件满足，被挂起的线程又会自动被唤醒）。

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

*阻塞添加     
	所谓阻塞添加，是指当阻塞队列元素已满时，队列会阻塞加入元素的线程，直到队列元素不满时才重新唤醒线程执行元素加入操作。
*阻塞删除    
	 阻塞删除是指在队列元素为空时，删除队列元素的线程将被阻塞，直到队列不为空时再执行删除操作(一般都会返回被删除的元素)。

 　　作为BlockingQueue的使用者，我们再也不用关心什么时候需要阻塞线程，什么时候需要唤醒线程，因为BlockingQueue把这一切都为我们包办了。

　　BlockingQueue的核心方法：

• 插入方法：

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

• 删除方法:

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

• 检查方法

  • element() ：获取但不移除此队列的头元素，没有元素则抛异常
  • peek() :获取但不移除此队列的头；若队列为空，则返回 null。

　　常见BlockingQueue：

　　①ArrayBlockingQueue

　　ArrayBlockingQueue是一个阻塞式的队列，继承自AbstractBlockingQueue,间接的实现了Queue接口和Collection接口。底层以数组的形式保存数据(实际上可看作一个循环数组)。常用的操作包括 add ,offer,put，remove,poll,take,peek。 前三者add offer put 是插入的操作。后面四个方法是取出的操作。　　

　　可以说，ArrayBlockingQueue 是一个用数组实现的有界阻塞队列，其内部按先进先出的原则对元素进行排序，其中put方法和take方法为添加和删除的阻塞方法。

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

　　②LinkedBlockingQueue

　　LinkedBlockingQueue是底层基于链表实现的阻塞队列，内部维持着一个数据缓冲队列（该队列由链表构成）。当生产者往队列中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者立即返回；只有当队列缓冲区达到最大值缓存容量时（LinkedBlockingQueue可以通过构造函数指定该值），才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生产者线程会被唤醒，反之对于消费者这端的处理也基于同样的原理。结构图如下：

　　LinkedBlockingQueue构造的时候若没有指定大小，则默认大小为Integer.MAX_VALUE，当然也可以在构造函数的参数中指定大小。LinkedBlockingQueue不接受null。
　　LinkedBlockingQueue之所以能够高效的处理并发数据，还因为其对于生产者端和消费者端分别采用了独立的锁来控制数据同步，这也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。

　　LinkedBlockingQueue中维持两把锁，一把锁用于入队，一把锁用于出队，这也就意味着，同一时刻，只能有一个线程执行入队，其余执行入队的线程将会被阻塞；同时，可以有另一个线程执行出队，其余执行出队的线程将会被阻塞。换句话说，虽然入队和出队两个操作同时均只能有一个线程操作，但是可以一个入队线程和一个出队线程共同执行，也就意味着可能同时有两个线程在操作队列，那么为了维持线程安全，LinkedBlockingQueue使用一个AtomicInterger类型的变量表示当前队列中含有的元素个数，所以可以确保两个线程之间操作底层队列是线程安全的。

　　LinkedBlockingQueue可以指定容量，内部维持一个队列，所以有一个头节点head和一个尾节点last，内部维持两把锁，一个用于入队，一个用于出队，还有锁关联的Condition对象。重要字段有：

    //容量，如果没有指定，该值为Integer.MAX_VALUE;
    private final int capacity;
    //当前队列中的元素
    private final AtomicInteger count = new AtomicInteger();
    //队列头节点，始终满足head.item==null
    transient Node<E> head;
    //队列的尾节点，始终满足last.next==null
    private transient Node<E> last;
    //用于出队的锁
    private final ReentrantLock takeLock = new ReentrantLock();
    //当队列为空时，保存执行出队的线程
    private final Condition notEmpty = takeLock.newCondition();
    //用于入队的锁
    private final ReentrantLock putLock = new ReentrantLock();
    //当队列满时，保存执行入队的线程
    private final Condition notFull = putLock.newCondition();

 　　LinkedBlockingQueue的构造方法有三个：

public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }

    public LinkedBlockingQueue(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node<E>(null);//last和head在队列为空时都存在，所以队列中至少有一个节点
    }

    public LinkedBlockingQueue(Collection<? extends E> c) {
        this(Integer.MAX_VALUE);
        final ReentrantLock putLock = this.putLock;
        putLock.lock(); // Never contended, but necessary for visibility
        try {
            int n = 0;
            for (E e : c) {
                if (e == null)
                    throw new NullPointerException();
                if (n == capacity)
                    throw new IllegalStateException("Queue full");
                enqueue(new Node<E>(e));
                ++n;
            }
            count.set(n);
        } finally {
            putLock.unlock();
        }
    }

 　　从LinkedBlockingQueue的构造方法中可以看出：当调用无参的构造方法时，容量是int的最大值；队列中至少包含一个节点，哪怕队列对外表现为空；LinkedBlockingQueue不支持null元素。

　　ArrayBlockingQueue和LinkedBlockingQueue是两个最普通、最常用的阻塞队列。
　　LinkedBlockingQueue用一个链表保存元素，其内部有一个Node的内部类，其中有一个成员变量 Node next，这样就形成了一个链表的结构，要获取下一个元素，只要调用next就可以了。而ArrayBlockingQueue则基于数组来保存元素。
　　LinkedBlockingQueue内部读写(插入获取)各有一个锁，而ArrayBlockingQueue则读写共享一个锁。

　　③SynchronousQueue

　　不像ArrayBlockingQueue或LinkedBlockingQueue，SynchronousQueue内部并没有数据缓存空间，你不能调用peek()方法来看队列中是否有数据元素，因为数据元素只有当你试着取走的时候才可能存在，不取走而只想偷窥一下是不行的，当然遍历这个队列的操作也是不允许的。队列头元素是第一个排队要插入数据的线程，而不是要交换的数据。数据是在配对的生产者和消费者线程之间直接传递的，并不会将数据缓冲到队列中。可以这样来理解：生产者和消费者互相等待对方，握手，然后一起离开。

　　SynchronousQueue的一个使用场景是在线程池里。Executors.newCachedThreadPool()就使用了SynchronousQueue，这个线程池根据需要（新任务到来时）创建新的线程，如果有空闲线程则会重复使用，线程空闲了60秒后会被回收。　

　　④LinkedBlockingDeque

　　LinkedBlockingDeque是一个基于链表的双端阻塞队列。和LinkedBlockingQueue类似，区别在于该类实现了Deque接口，而LinkedBlockingQueue实现了Queue接口。

　　LinkedBlockingDeque是一个可选容量的阻塞队列，如果没有设置容量，那么容量将是Int的最大值。

　　LinkedBlockingDeque的底层数据结构是一个双端队列，该队列使用链表实现，如图所示：

　

　　LinkedBlockingDeque的重要字段有如下几个：

    //队列的头节点
    transient Node<E> first;
    //队列的尾节点
    transient Node<E> last;
    //队列中元素的个数
    private transient int count;
    //队列中元素的最大个数
    private final int capacity;
    //锁
    final ReentrantLock lock = new ReentrantLock();
    //队列为空时，阻塞take线程的条件队列
    private final Condition notEmpty = lock.newCondition();
    //队列满时，阻塞put线程的条件队列
    private final Condition notFull = lock.newCondition();

 　　从上面的字段，可以看到LinkedBlockingDeque内部只有一把锁以及该锁上关联的两个条件，所以可以推断同一时刻只有一个线程可以在队头或者队尾执行入队或出队操作。可以发现这点和LinkedBlockingQueue不同，LinkedBlockingQueue可以同时有两个线程在两端执行操作。

　　由于LinkedBlockingDeque是一个双端队列，所以就可以在队头执行入队和出队操作，也可以在队尾执行入队和出队操作。

　　LinkedBlockingDeque的构造方法有三个：

public LinkedBlockingDeque() {
        this(Integer.MAX_VALUE);
    }

    public LinkedBlockingDeque(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
    }

    public LinkedBlockingDeque(Collection<? extends E> c) {
        this(Integer.MAX_VALUE);
        final ReentrantLock lock = this.lock;
        lock.lock(); // Never contended, but necessary for visibility
        try {
            for (E e : c) {
                if (e == null)
                    throw new NullPointerException();
                if (!linkLast(new Node<E>(e)))
                    throw new IllegalStateException("Deque full");
            }
        } finally {
            lock.unlock();
        }
    }

 　　可以看到这三个构造方法的结构和LinkedBlockingQueue是相同的。 但是LinkedBlockingQueue是存在一个哨兵节点维持头节点的，而LinkedBlockingDeque中是没有的。　　

LinkedBlockingDeque和LinkedBlockingQueue的相同点在于：
	①基于链表
	②容量可选，不设置的话，就是Int的最大值
LinkedBlockingDeque和LinkedBlockingQueue的不同点在于：
	①双端链表和单链表
	②不存在哨兵节点
	③一把锁+两个条件

LinkedBlockingDeque和ArrayBlockingQueue的相同点在于：使用一把锁+两个条件维持队列的同步。

ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
final int index = i;
singleThreadExecutor.execute(new Runnable() {
 
@Override
public void run() {
try {
System.out.println(index);
Thread.sleep(2000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
}