所有分类
  • 所有分类
  • 未分类

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. 任务1和2在核心线程中执行;
  2. 任务3和4进来时,放到ArrayBlockingQueue缓存队列中,并且只能放2个(ArrayBlockingQueue设置的大小为2);
  3. 任务5和6进来的时候,任务5新建线程来执行任务,已经达到最大线程数3,所以任务6拒绝;
  4. 当有线程执行完的时候,再将任务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的相同点:

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

不同点

LinkedBlockingQueueArrayBlockingQueue
大小指定实例化时可指定队列大小。 若不指定大小,会采用默认的Integer.MAX_VALUE。实例化时必须指定大小。
吞吐量大。 采用了“双锁队列” 算法,元素的入队和出队分别由putLock、takeLock两个独立的可重入锁来实现。小。 几乎每一个方法都需要先获取同一个ReentrantLock独占锁才能进行。

实例

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

  1. 核心线程执行任务1和2,其它的任务3~6放到队列中
  2. 执行完1和2,将3和4从队列中取出执行
  3. 执行完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个任务:

  1. 2个核心线程执行2个任务;
  2. 第3个任务的时候,创建线程来执行任务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的内部实现

顺序消费实例

见:Java多线程–使用阻塞队列实现顺序消费–方法/实例 – 自学精灵

0

评论0

请先

显示验证码
没有账号?注册  忘记密码?

社交账号快速登录