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从队列中取出执行

创建线程池代码如下：

代码语言：javascript复制

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

执行结果如下：

代码语言：javascript复制

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从队列中取出

创建线程池代码如下：

代码语言：javascript复制

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

运行结果如下：

代码语言：javascript复制

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，所以拒绝任务。

代码：

代码语言：javascript复制

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

执行结果：

代码语言：javascript复制

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的内部实现

参考文献： https://knife.blog.csdn.net/article/details/122795966

java 队列多线程排序线程

0 人点赞