BlockingQueue 阻塞队列

时间 2019-11-11

原文原文链接

阻塞队列，顾名思义，首先它是一个队列，而一个队列在数据结构中所起的做用大体以下图所示：java

从上图咱们能够很清楚看到，经过一个共享的队列，可使得数据由队列的一端输入，从另一端输出；
经常使用的队列主要有如下两种：（固然经过不一样的实现方式，还能够延伸出不少不一样类型的队列，DelayQueue就是其中的一种）程序员

先进先出（FIFO）：先插入的队列的元素也最早出队列，相似于排队的功能。从某种程度上来讲这种队列也体现了一种公平性。
后进先出（LIFO）：后插入队列的元素最早出队列，这种队列优先处理最近发生的事件。

多线程环境中，经过队列能够很容易实现数据共享，好比经典的“生产者”和“消费者”模型中，经过队列能够很便利地实现二者之间的数据共享。假设咱们有若干生产者线程，另外又有若干个消费者线程。若是生产者线程须要把准备好的数据共享给消费者线程，利用队列的方式来传递数据，就能够很方便地解决他们之间的数据共享问题。但若是生产者和消费者在某个时间段内，万一发生数据处理速度不匹配的状况呢？理想状况下，若是生产者产出数据的速度大于消费者消费的速度，而且当生产出来的数据累积到必定程度的时候，那么生产者必须暂停等待一下（阻塞生产者线程），以便等待消费者线程把累积的数据处理完毕，反之亦然。然而，在concurrent包发布之前，在多线程环境下，咱们每一个程序员都必须去本身控制这些细节，尤为还要兼顾效率和线程安全，而这会给咱们的程序带来不小的复杂度。好在此时，强大的concurrent包横空出世了，而他也给咱们带来了强大的BlockingQueue。（在多线程领域：所谓阻塞，在某些状况下会挂起线程（即阻塞），一旦条件知足，被挂起的线程又会自动被唤醒）数组

BlockingQueue的核心方法：
放入数据：缓存

offer(anObject):表示若是可能的话,将anObject加到BlockingQueue里,即若是BlockingQueue能够容纳,则返回true,不然返回false.（本方法不阻塞当前执行方法的线程）
offer(E o, long timeout, TimeUnit unit),能够设定等待的时间，若是在指定的时间内，还不能往队列中加入BlockingQueue，则返回失败。
put(anObject):把anObject加到BlockingQueue里,若是BlockQueue没有空间,则调用此方法的线程被阻断直到BlockingQueue里面有空间再继续.

获取数据：安全

poll(time):取走BlockingQueue里排在首位的对象,若不能当即取出,则能够等time参数规定的时间,取不到时返回null;
poll(long timeout, TimeUnit unit)：从BlockingQueue取出一个队首的对象，若是在指定时间内，队列一旦有数据可取，则当即返回队列中的数据。不然知道时间超时尚未数据可取，返回失败。
take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的数据被加入;
drainTo():一次性从BlockingQueue获取全部可用的数据对象（还能够指定获取数据的个数），经过该方法，能够提高获取数据效率；不须要屡次分批加锁或释放锁。

BlockingQueue 具备 4 组不一样的方法用于插入、移除以及对队列中的元素进行检查。若是请求的操做不能获得当即执行的话，每一个方法的表现也不一样。这些方法以下：数据结构

	抛异常	特定值	阻塞	超时
插入	add(o)	offer(o)	put(o)	offer(o, timeout, timeunit)
移除	remove(o)	poll(o)	take(o)	poll(timeout, timeunit)
检查	element(o)	peek(o)

四组不一样的行为方式解释：多线程

抛异常：若是试图的操做没法当即执行，抛一个异常。
特定值：若是试图的操做没法当即执行，返回一个特定的值(经常是 true / false)。
阻塞：若是试图的操做没法当即执行，该方法调用将会发生阻塞，直到可以执行。
超时：若是试图的操做没法当即执行，该方法调用将会发生阻塞，直到可以执行，但等待时间不会超过给定值。返回一个特定值以告知该操做是否成功(典型的是 true / false)。

没法向一个 BlockingQueue 中插入 null。若是你试图插入 null，BlockingQueue 将会抛出一个 NullPointerException。
能够访问到 BlockingQueue 中的全部元素，而不只仅是开始和结束的元素。好比说，你将一个对象放入队列之中以等待处理，但你的应用想要将其取消掉。那么你能够调用诸如 remove(o) 方法来将队列之中的特定对象进行移除。可是这么干效率并不高(译者注：基于队列的数据结构，获取除开始或结束位置的其余对象的效率不会过高)，所以你尽可能不要用这一类的方法，除非你确实不得不那么作。并发

常见BlockingQueue
在了解了BlockingQueue的基本功能后，让咱们来看看BlockingQueue家庭大体有哪些成员？ dom

BlockingQueue成员详细介绍ide

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），这样的话，若是生产者的速度一旦大于消费者的速度，也许尚未等到队列满阻塞产生，系统内存就有可能已被消耗殆尽了。

ArrayBlockingQueue和LinkedBlockingQueue是两个最普通也是最经常使用的阻塞队列，通常状况下，在处理多线程间的生产者消费者问题，使用这两个类足以。

下面的代码演示了如何使用BlockingQueue：

public class BlockingQueueTest {

    /**
     * @param args the command line arguments
     */
    public static void main(String[] args) {
        // 声明一个容量为10的缓存队列
        BlockingQueue sharedQueue = new LinkedBlockingQueue(10);

        //Creating Producer and Consumer Thread
        Thread prodThread = new Thread(new Producer(sharedQueue));
        Thread consThread = new Thread(new Consumer(sharedQueue));
        Thread consThread1 = new Thread(new Consumer(sharedQueue));

        //Starting producer and Consumer thread
        prodThread.start();
        consThread.start();
        consThread1.start();
    }

}

/**
 * 生产者线程
 */
public class Producer implements Runnable {
 
    private final BlockingQueue sharedQueue;
 
    public Producer(BlockingQueue sharedQueue) {
        this.sharedQueue = sharedQueue;
    }
 
    @Override
    public void run() {
        for(int i=0; i<10; i++){
            try {
                System.out.println("Produced: " + i);
                sharedQueue.put(i);
            } catch (InterruptedException ex) {
                System.out.println(ex);
            }
        }
    }
 
}

/**
 * 消费者线程
 */
public class Consumer implements Runnable {

    private final BlockingQueue sharedQueue;

    public Consumer(BlockingQueue sharedQueue) {
        this.sharedQueue = sharedQueue;
    }

    @Override
    public void run() {
        while (true) {
            try {
                System.out.println(Thread.currentThread().getName()+"Consumed: " + sharedQueue.take());
            } catch (InterruptedException ex) {
                System.out.println(ex);
            }
        }
    }

}

output:

Produced: 0
Produced: 1
Produced: 2
Produced: 3
Produced: 4
Produced: 5
Produced: 6
Produced: 7
Produced: 8
Produced: 9
Thread-1Consumed: 0
Thread-1Consumed: 2
Thread-1Consumed: 3
Thread-1Consumed: 4
Thread-1Consumed: 5
Thread-1Consumed: 6
Thread-1Consumed: 7
Thread-1Consumed: 8
Thread-1Consumed: 9
Thread-2Consumed: 1

3. DelayQueue
DelayQueue中的元素只有当其指定的延迟时间到了，才可以从队列中获取到该元素。DelayQueue是一个没有大小限制的队列，所以往队列中插入数据的操做（生产者）永远不会被阻塞，而只有获取数据的操做（消费者）才会被阻塞。
使用场景：
DelayQueue使用场景较少，但都至关巧妙，常见的例子好比使用一个DelayQueue来管理一个超时未响应的链接队列。

4. PriorityBlockingQueue
基于优先级的阻塞队列（优先级的判断经过构造函数传入的Compator对象来决定），但须要注意的是PriorityBlockingQueue并不会阻塞数据生产者，而只会在没有可消费的数据时，阻塞数据的消费者。所以使用的时候要特别注意，生产者生产数据的速度绝对不能快于消费者消费数据的速度，不然时间一长，会最终耗尽全部的可用堆内存空间。在实现PriorityBlockingQueue时，内部控制线程同步的锁采用的是公平锁。

5. SynchronousQueue
一种无缓冲的等待队列，相似于无中介的直接交易，它的内部同时只可以容纳单个元素。若是该队列已有一元素的话，试图向队列中插入一个新元素的线程将会阻塞，直到另外一个线程将该元素从队列中抽走。一样，若是该队列为空，试图向队列中抽取一个元素的线程将会阻塞，直到另外一个线程向队列中插入了一条新的元素。

有点像原始社会中的生产者和消费者，生产者拿着产品去集市销售给产品的最终消费者，而消费者必须亲自去集市找到所要商品的直接生产者，若是一方没有找到合适的目标，那么对不起，你们都在集市等待。相对于有缓冲的BlockingQueue来讲，少了一个中间经销商的环节（缓冲区），若是有经销商，生产者直接把产品批发给经销商，而无需在乎经销商最终会将这些产品卖给那些消费者，因为经销商能够库存一部分商品，所以相对于直接交易模式，整体来讲采用中间经销商的模式会吞吐量高一些（能够批量买卖）；但另外一方面，又由于经销商的引入，使得产品从生产者到消费者中间增长了额外的交易环节，单个产品的及时响应性能可能会下降。

声明一个SynchronousQueue有两种不一样的方式，它们之间有着不太同样的行为。公平模式和非公平模式的区别:
　　若是采用公平模式：SynchronousQueue会采用公平锁，并配合一个FIFO队列来阻塞多余的生产者和消费者，从而体系总体的公平策略；
　　但若是是非公平模式（SynchronousQueue默认）：SynchronousQueue采用非公平锁，同时配合一个LIFO队列来管理多余的生产者和消费者，然后一种模式，若是生产者和消费者的处理速度有差距，则很容易出现饥渴的状况，便可能有某些生产者或者是消费者的数据永远都得不处处理。

下面使用SynchronousQueue模拟只能生产一个产品的生产者-消费者模型：

public class SynchronousQueueTest {

    public static void main(String[] args) {
        SynchronousQueue<Integer> queue = new SynchronousQueue<>();
        new Thread(new Customer(queue)).start();
        new Thread(new Product(queue)).start();
    }
}

public class Product implements Runnable {

    SynchronousQueue<Integer> queue;

    public Product(SynchronousQueue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while (true) {
            int rand = new Random().nextInt(1000);
            System.out.println("生产了一个产品：" + rand);
            System.out.println("等待三秒后运送出去...");
            try {
                TimeUnit.SECONDS.sleep(3);
                queue.put(rand);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

public class Customer implements Runnable {

    SynchronousQueue<Integer> queue;

    public Customer(SynchronousQueue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while (true) {
            try {
                System.out.println("消费了一个产品:" + queue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("------------------------------------------");
        }
    }

}

output:

生产了一个产品：827
等待三秒后运送出去...
消费了一个产品:827
------------------------------------------
生产了一个产品：561
等待三秒后运送出去...
消费了一个产品:561
------------------------------------------
生产了一个产品：392
等待三秒后运送出去...
消费了一个产品:392
------------------------------------------
生产了一个产品：790
等待三秒后运送出去...
消费了一个产品:790
------------------------------------------
生产了一个产品：380

小结：　　BlockingQueue不光实现了一个完整队列所具备的基本功能，同时在多线程环境下，他还自动管理了多线间的自动等待于唤醒功能，从而使得程序员能够忽略这些细节，关注更高级的功能。