Java多线程系列--阻塞队列(BlockingQueue)的用法(有实例)

简介

说明

        本文用示例介绍Java中阻塞队列（BlockingQueue）的用法。

队列类型

        BlockingQueue有这几种类型：ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue、PriorityBlockingQueue、DelayedWorkQueue。

队列类型
    

说明

ArrayBlockingQueue
    

        基于数组的FIFO队列；有界；创建时必须指定大小；

        入队和出队共用一个可重入锁。默认使用非公平锁。

LinkedBlockingQueue
    

        基于链表的FIFO队列；有/无界；默认大小是 Integer.MAX_VALUE（无界），可自定义（有界）；

        两个重入锁分别控制元素的入队和出队，用Condition进行线程间的唤醒和等待。

        吞吐量通常要高于ArrayBlockingQueue。

        默认大小的LinkedBlockingQueue将导致所有 corePoolSize 线程都忙时新任务在队列中等待。这样，创建的线程不会超过 corePoolSize。（因此，maximumPoolSize 的值也就无效了）。当每个任务相互独时，适合使用无界队列；例如， 在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。

SynchronousQueue
    

        无缓存的等待队列；无界；可认为大小为0。
        不保存提交任务，直接提交出去。若超出corePoolSize个任务，直接创建新线程来执行任务，直到(corePoolSize＋新建线程)> maximumPoolSize。

        此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线 程具有增长的可能性。

        吞吐量通常要高于LinkedBlockingQueue。

        //也有地方说：是一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态

        详见下边的：CachedThreadPool的execute流程

PriorityBlockingQueue
    

        基于链表的优先级队列；有/无界；默认大小是 Integer.MAX_VALUE，可自定义；

        类似于LinkedBlockingQueue，但是其所含对象的排序不是FIFO，而是依据对象的自然顺序或者构造函数的Comparator决定。

DelayedWorkQueue
    

常用方法

放入数据

方法
    

说明

offer(E e)
    

向队列尾部插入一个元素。该方法是非阻塞的。

如果队列中有空闲：插入成功后返回 true。

如果队列己满：丢弃当前元素然后返回false。

如果e元素为null：抛出NullPointerException异常。

offer(E o, long timeout, TimeUnit unit)
    

可以设定等待的时间，若在指定的时间内，还不能往队列中加入BlockingQueue，则返回失败。

add(E e)
    

内部调用offer方法。

与直接调用offer的区别：

add：失败时，抛出异常

offer：失败时，返回false

put(E e)
    

向队列尾部插入一个元素。

如果队列中有空闲：插入后直接返回。

如果队列己满：阻塞当前线程，直到队列有空闲插入成功后返回。

如果在阻塞时被其他线程设置了中断标志：被阻塞线程会抛出InterruptedException异常而返回。

如果e元素为null：抛出NullPointerException异常。

获取数据

方法
    

说明

poll()
    

获取当前队列头部元素并从队列里面移除它。

如果队列为空则返回null。

poll(long timeout, TimeUnit unit)
    

从BlockingQueue取出（会删除对象）一个队首的对象。

一旦在指定时间内有数据可取，则立即返回队列中的数据。

若直到时间超时还没有数据可取，返回失败。

take()
    

获取当前队列头部元素并从队列里面移除它。

如果队列为空则阻塞当前线程直到队列不为空然后返回元素；

如果在阻塞时被其他线程设置了中断标志，则被阻塞线程会抛出InterruptedException异常而返回。

drainTo()
    

一次性从BlockingQueue获取（会删除对象）所有可用的数据对象（可指定获取数据的个数）。

本方法可提升获取数据效率，不需要多次分批加锁或释放锁。

其他方法

方法
    

说明

remainingCapacity()
    

获取队列中剩余的空间

contains(Object o)
    

判断队列中是否拥有该值。

remove(Object o)
    

从队列中移除指定的值。

size()
    

获得队列中有多少值。

（返回AtomicLong的值）
ArrayBlockingQueue

简介

        ArrayBlockingQueue通过数组实现的FIFO有界阻塞队列，它的大小在实例被初始化的时候就被固定了，不能更改。

        该类支持一个可选的公平策略，用于被阻塞等待的线程获取独占锁的排序，因为ArrayBlockingQueue内部的操作都需要获取一个ReentrantLock锁，该锁是支持公平策略的，所以ArrayBlockingQueue的公平策略就直接作用于ReentrantLock锁，决定线程是否有公平获取锁的权利。默认情况下是非公平的，公平模式下队列按照FIFO顺序授予线程访问权。公平性通常会降低吞吐量，但会降低可变性并避免饥饿。

ArrayBlockingQueue的缺陷

        通过源码可以看见，ArrayBlockingQueue内部的几乎每一个操作方法都需要先获取同一个ReentrantLock独占锁才能进行，这极大的降低了吞吐量，几乎每个操作都会阻塞其它操作，最主要是插入操作和取出操作相互之间互斥。所以ArrayBlockingQueue不适用于需要高吞吐量的高效率数据生成与消费场景。LinkedBlockingQueue就能弥补其低吞吐量的缺陷。

实例

创建一个corePoolSize为2，maximumPoolSize为3的线程池。其中ArrayBlockingQueue设置缓存2个任务。执行6个任务。ArrayBlockingQueue为有界队列：

    任务1和2在核心线程中执行；
    任务3和4进来时，放到ArrayBlockingQueue缓存队列中，并且只能放2个（ArrayBlockingQueue设置的大小为2）；
    任务5和6进来的时候，任务5新建线程来执行任务，已经达到最大线程数3，所以任务6拒绝；
    当有线程执行完的时候，再将任务3和4从队列中取出执行

创建线程池代码如下：

      /**
         * ArrayBlockingQueue
         */
        private static void arrayQueue() {
            System.out.println("\n\n =======ArrayBlockingQueue====== \n\n");
            Executor executors = new ThreadPoolExecutor(
                    2, 3, 30, TimeUnit.SECONDS,
                    new ArrayBlockingQueue<Runnable>(2),
                    new RejectHandler());
            execute(executors);
        }

 执行结果如下

    1 is running...
    2 is running...
    6 is rejected ^^ //6被拒
    5 is running...  //5新建线程执行
    1 is end !!!
    2 is end !!!
    3 is running... //1和2执行完之后3和4才执行
    4 is running...
    5 is end !!!

LinkedBlockingQueue

简介

LinkedBlockingQueue和ArrayBlockingQueue的相同点：

    是FIFO队列，不允许插入null值。
    容量在实例被构造完成之后不允许被更改

不同点

项
    

LinkedBlockingQueue
    

ArrayBlockingQueue

大小指定
    

实例化时可指定队列大小。

若不指定大小，会采用默认的Integer.MAX_VALUE。
    

实例化时必须指定大小。

吞吐量
    

大。

采用了“双锁队列” 算法，元素的入队和出队分别由putLock、takeLock两个独立的可重入锁来实现。
    

小。

几乎每一个方法都需要先获取同一个ReentrantLock独占锁才能进行。

实例

创建一个corePoolSize为2，maximumPoolSize为3的线程池。无界队列。同样执行6个任务

    核心线程执行任务1和2，其它的任务3～6放到队列中
    执行完1和2，将3和4从队列中取出执行
    执行完3和4，将5和6从队列中取出

创建线程池代码如下：

    /**
     * LinkedBlockingQueue
     */
    private static void linkedQueue() {
        System.out.println("\n\n =======LinkedBlockingQueue====== \n\n");
        Executor executors = new ThreadPoolExecutor(
                2, 3, 30, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(),
                new RejectHandler());
        execute(executors);
    }

运行结果如下：

    1 is running...
    2 is running... //中间线程休眠
    2 is end !!!   //10s之后才运行完
    1 is end !!!
    3 is running...  //任务3和4才执行
    4 is running...
    4 is end !!!
    3 is end !!!
    6 is running...
    5 is running...
    5 is end !!!
    6 is end !!!

SynchronousQueue

说明

        SynchrousQueue是个一个无缓存的队列。因为：SynchrousQueue源码可以看到：isEmpty()始终为true；size()始终返回0。

示例

说明

        创建一个corePoolSize为2，maximumPoolSize为3的线程池。执行6个任务。

根据参数设置应该只可以执行3个任务：

    2个核心线程执行2个任务；
    第3个任务的时候，创建线程来执行任务3；
    当第4个任务来的时候，此时已经超过了maximumPoolSize，所以拒绝任务。

代码

    /**
    * SynchronousQueue
    */
    private static void syncQueue() {
        System.out.println("\n\n =======SynchronousQueue====== \n\n");
        Executor executors = new ThreadPoolExecutor(
                2, 3, 30, TimeUnit.SECONDS,
                new SynchronousQueue<Runnable>(),
                new RejectHandler());
        execute(executors);
    }

执行结果

    1 is running...
    4 is rejected ^^ //4被拒
    2 is running...
    3 is running...
    5 is rejected ^^  //5被拒
    6 is rejected ^^  //6被拒
    3 is end !!!
    1 is end !!!
    2 is end !!!

PriorityBlockingQueue

简介

        PriorityBlockingQueue是一个无限容量的阻塞队列。

        容量是无限的，所以put等入队操作其实不存在阻塞，只要内存足够都能够立即入队成功，当然多个入队操作的线程之间还是存在竞争唯一锁的互斥访问。虽然PriorityBlockingQueue逻辑上是无界的，但是尝试添加元素时还是可能因为资源耗尽而抛出OutOfMemoryError。

        该队列也不允许放入null值，它使用与类java.util.PriorityQueue 相同的排序规则，也不允许放入不可比较的对象，这样做会导致ClassCastException。

        值得注意的是，虽然PriorityBlockingQueue叫优先级队列，但是并不是说元素一入队就会按照排序规则被排好序，而是只有通过调用take、poll方法出队或者drainTo转移出的队列顺序才是被优先级队列排过序的。所以通过调用 iterator() 以及可拆分迭代器 spliterator() 方法返回的迭代器迭代的元素顺序都没有被排序。如果需要有序遍历可以通过 Arrays.sort(pq.toArray()) 方法来排序。注意peek方法永远只获取且不删除第一个元素，所以多次调用peek都是返回同样的值。

        PriorityBlockingQueue其实是通过Comparator来排序的，要么入队的元素实现了Comparator接口（即所谓的自然排序），要么构造PriorityBlockingQueue实例的时候传入一个统一的Comparator实例，如果两者兼备那么以后者为准

        PriorityBlockingQueue不保证具有相同优先级的元素顺序，但是你可以定义自定义类或比较器，通过辅助属性来决定优先级相同的元素的顺序，后文会举例说明。
DelayedWorkQueue

简介

        为什么不直接使用DelayQueue而要重新实现一个DelayedWorkQueue呢，可能是了方便在实现过程中加入一些扩展。

使用场景

    实现重试机制

        比如当调用接口失败后，把当前调用信息放入delay=10s的元素，然后把元素放入队列，那么这个队列就是一个重试队列。一个线程通过take方法获取需要重试的接口，take返回则接口进行重试，失败则再次放入队列，同时也可以在元素加上重试次数。

    TimerQueue的内部实现