这是java高并发系列第25篇文章。

环境:jdk1.8。

  1. 掌握Queue、BlockingQueue接口中常用的方法
  2. 介绍6中阻塞队列,及相关场景示例
  3. 重点掌握4种常用的阻塞队列

队列是一种先进先出(FIFO)的数据结构,java中用Queue接口来表示队列。

Queue接口中定义了6个方法:

  1. public interface Queue<E> extends Collection<E> {
  2. boolean add(e);
  3. boolean offer(E e);
  4. E remove();
  5. E poll();
  6. E element();
  7. E peek();
  8. }

每个Queue方法都有两种形式:

(1)如果操作失败则抛出异常,

(2)如果操作失败,则返回特殊值(nullfalse,具体取决于操作),接口的常规结构如下表所示。

操作类型 抛出异常 返回特殊值
插入 add(e) offer(e)
移除 remove() poll()
检查 element() peek()

QueueCollection继承的add方法插入一个元素,除非它违反了队列的容量限制,在这种情况下它会抛出IllegalStateExceptionoffer方法与add不同之处仅在于它通过返回false来表示插入元素失败。

removepoll方法都移除并返回队列的头部,确切地移除哪个元素是由具体的实现来决定的,仅当队列为空时,removepoll方法的行为才有所不同,在这些情况下,remove抛出NoSuchElementException,而poll返回null

elementpeek方法返回队列头部的元素,但不移除,它们之间的差异与removepoll的方式完全相同,如果队列为空,则element抛出NoSuchElementException,而peek返回null

队列一般不要插入空元素。

BlockingQueue位于juc中,熟称阻塞队列, 阻塞队列首先它是一个队列,继承Queue接口,是队列就会遵循先进先出(FIFO)的原则,又因为它是阻塞的,故与普通的队列有两点区别:

  1. 当一个线程向队列里面添加数据时,如果队列是满的,那么将阻塞该线程,暂停添加数据
  2. 当一个线程从队列里面取出数据时,如果队列是空的,那么将阻塞该线程,暂停取出数据

BlockingQueue相关方法:

操作类型 抛出异常 返回特殊值 一直阻塞 超时退出
插入 add(e) offer(e) put(e) offer(e,timeuout,unit)
移除 remove() poll() take() poll(timeout,unit)
检查 element() peek() 不支持 不支持

重点,再来解释一下,加深印象:

  1. 3个可能会有异常的方法,add、remove、element;这3个方法不会阻塞(是说队列满或者空的情况下是否会阻塞);队列满的情况下,add抛出异常;队列为空情况下,remove、element抛出异常
  2. offer、poll、peek 也不会阻塞(是说队列满或者空的情况下是否会阻塞);队列满的情况下,offer返回false;队列为空的情况下,pool、peek返回null
  3. 队列满的情况下,调用put方法会导致当前线程阻塞
  4. 队列为空的情况下,调用take方法会导致当前线程阻塞
  5. offer(e,timeuout,unit),超时之前,插入成功返回true,否者返回false
  6. poll(timeout,unit),超时之前,获取到头部元素并将其移除,返回true,否者返回false
  7. 以上一些方法希望大家都记住,方便以后使用

看一下相关类图

ArrayBlockingQueue

基于数组的阻塞队列实现,其内部维护一个定长的数组,用于存储队列元素。线程阻塞的实现是通过ReentrantLock来完成的,数据的插入与取出共用同一个锁,因此ArrayBlockingQueue并不能实现生产、消费同时进行。而且在创建ArrayBlockingQueue时,我们还可以控制对象的内部锁是否采用公平锁,默认采用非公平锁。

LinkedBlockingQueue

基于单向链表的阻塞队列实现,在初始化LinkedBlockingQueue的时候可以指定大小,也可以不指定,默认类似一个无限大小的容量(Integer.MAX_VALUE),不指队列容量大小也是会有风险的,一旦数据生产速度大于消费速度,系统内存将有可能被消耗殆尽,因此要谨慎操作。另外LinkedBlockingQueue中用于阻塞生产者、消费者的锁是两个(锁分离),因此生产与消费是可以同时进行的。

PriorityBlockingQueue

一个支持优先级排序的无界阻塞队列,进入队列的元素会按照优先级进行排序

SynchronousQueue

同步阻塞队列,SynchronousQueue没有容量,与其他BlockingQueue不同,SynchronousQueue是一个不存储元素的BlockingQueue,每一个put操作必须要等待一个take操作,否则不能继续添加元素,反之亦然

DelayQueue

DelayQueue是一个支持延时获取元素的无界阻塞队列,里面的元素全部都是“可延期”的元素,列头的元素是最先“到期”的元素,如果队列里面没有元素到期,是不能从列头获取元素的,哪怕有元素也不行,也就是说只有在延迟期到时才能够从队列中取元素

LinkedTransferQueue

LinkedTransferQueue是基于链表的FIFO无界阻塞队列,它出现在JDK7中,Doug Lea 大神说LinkedTransferQueue是一个聪明的队列,它是ConcurrentLinkedQueue、SynchronousQueue(公平模式下)、无界的LinkedBlockingQueues等的超集,LinkedTransferQueue包含了ConcurrentLinkedQueue、SynchronousQueue、LinkedBlockingQueues三种队列的功能

下面我们来介绍每种阻塞队列的使用。

有界阻塞队列,内部使用数组存储元素,有2个常用构造方法:

  1. //capacity表示容量大小,默认内部采用非公平锁
  2. public ArrayBlockingQueue(int capacity)
  3. //capacity:容量大小,fair:内部是否是使用公平锁
  4. public ArrayBlockingQueue(int capacity, boolean fair)

需求:业务系统中有很多地方需要推送通知,由于需要推送的数据太多,我们将需要推送的信息先丢到阻塞队列中,然后开一个线程进行处理真实发送,代码如下:

  1. package com.itsoku.chat25;
  2. import lombok.Data;
  3. import lombok.extern.slf4j.Slf4j;
  4. import sun.text.normalizer.NormalizerBase;
  5. import java.util.Calendar;
  6. import java.util.concurrent.*;
  7. /**
  8. * 跟着阿里p7学并发,微信公众号:javacode2018
  9. */
  10. public class Demo1 {
  11. //推送队列
  12. static ArrayBlockingQueue<String> pushQueue = new ArrayBlockingQueue<String>(10000);
  13. static {
  14. //启动一个线程做真实推送
  15. new Thread(() -> {
  16. while (true) {
  17. String msg;
  18. try {
  19. long starTime = System.currentTimeMillis();
  20. //获取一条推送消息,此方法会进行阻塞,直到返回结果
  21. msg = pushQueue.take();
  22. long endTime = System.currentTimeMillis();
  23. //模拟推送耗时
  24. TimeUnit.MILLISECONDS.sleep(500);
  25. System.out.println(String.format("[%s,%s,take耗时:%s],%s,发送消息:%s", starTime, endTime, (endTime - starTime), Thread.currentThread().getName(), msg));
  26. } catch (InterruptedException e) {
  27. e.printStackTrace();
  28. }
  29. }
  30. }).start();
  31. }
  32. //推送消息,需要发送推送消息的调用该方法,会将推送信息先加入推送队列
  33. public static void pushMsg(String msg) throws InterruptedException {
  34. pushQueue.put(msg);
  35. }
  36. public static void main(String[] args) throws InterruptedException {
  37. for (int i = 1; i <= 5; i++) {
  38. String msg = "一起来学java高并发,第" + i + "天";
  39. //模拟耗时
  40. TimeUnit.SECONDS.sleep(i);
  41. Demo1.pushMsg(msg);
  42. }
  43. }
  44. }

输出:

  1. [1565595629206,1565595630207,take耗时:1001],Thread-0,发送消息:一起来学java高并发,第1
  2. [1565595630208,1565595632208,take耗时:2000],Thread-0,发送消息:一起来学java高并发,第2
  3. [1565595632208,1565595635208,take耗时:3000],Thread-0,发送消息:一起来学java高并发,第3
  4. [1565595635208,1565595639209,take耗时:4001],Thread-0,发送消息:一起来学java高并发,第4
  5. [1565595639209,1565595644209,take耗时:5000],Thread-0,发送消息:一起来学java高并发,第5

代码中我们使用了有界队列ArrayBlockingQueue,创建ArrayBlockingQueue时候需要制定容量大小,调用pushQueue.put将推送信息放入队列中,如果队列已满,此方法会阻塞。代码中在静态块中启动了一个线程,调用pushQueue.take();从队列中获取待推送的信息进行推送处理。

注意:ArrayBlockingQueue如果队列容量设置的太小,消费者发送的太快,消费者消费的太慢的情况下,会导致队列空间满,调用put方法会导致发送者线程阻塞,所以注意设置合理的大小,协调好消费者的速度。

内部使用单向链表实现的阻塞队列,3个构造方法:

  1. //默认构造方法,容量大小为Integer.MAX_VALUE
  2. public LinkedBlockingQueue();
  3. //创建指定容量大小的LinkedBlockingQueue
  4. public LinkedBlockingQueue(int capacity);
  5. //容量为Integer.MAX_VALUE,并将传入的集合丢入队列中
  6. public LinkedBlockingQueue(Collection<? extends E> c);

LinkedBlockingQueue的用法和ArrayBlockingQueue类似,建议使用的时候指定容量,如果不指定容量,插入的太快,移除的太慢,可能会产生OOM。

无界的优先级阻塞队列,内部使用数组存储数据,达到容量时,会自动进行扩容,放入的元素会按照优先级进行排序,4个构造方法:

  1. //默认构造方法,默认初始化容量是11
  2. public PriorityBlockingQueue();
  3. //指定队列的初始化容量
  4. public PriorityBlockingQueue(int initialCapacity);
  5. //指定队列的初始化容量和放入元素的比较器
  6. public PriorityBlockingQueue(int initialCapacity,Comparator<? super E> comparator);
  7. //传入集合放入来初始化队列,传入的集合可以实现SortedSet接口或者PriorityQueue接口进行排序,如果没有实现这2个接口,按正常顺序放入队列
  8. public PriorityBlockingQueue(Collection<? extends E> c);

优先级队列放入元素的时候,会进行排序,所以我们需要指定排序规则,有2种方式:

  1. 创建PriorityBlockingQueue指定比较器Comparator
  2. 放入的元素需要实现Comparable接口

上面2种方式必须选一个,如果2个都有,则走第一个规则排序。

需求:还是上面的推送业务,目前推送是按照放入的先后顺序进行发送的,比如有些公告比较紧急,优先级比较高,需要快点发送,怎么搞?此时PriorityBlockingQueue就派上用场了,代码如下:

  1. package com.itsoku.chat25;
  2. import java.util.concurrent.PriorityBlockingQueue;
  3. import java.util.concurrent.TimeUnit;
  4. /**
  5. * 跟着阿里p7学并发,微信公众号:javacode2018
  6. */
  7. public class Demo2 {
  8. //推送信息封装
  9. static class Msg implements Comparable<Msg> {
  10. //优先级,越小优先级越高
  11. private int priority;
  12. //推送的信息
  13. private String msg;
  14. public Msg(int priority, String msg) {
  15. this.priority = priority;
  16. this.msg = msg;
  17. }
  18. @Override
  19. public int compareTo(Msg o) {
  20. return Integer.compare(this.priority, o.priority);
  21. }
  22. @Override
  23. public String toString() {
  24. return "Msg{" +
  25. "priority=" + priority +
  26. ", msg='" + msg + '\'' +
  27. '}';
  28. }
  29. }
  30. //推送队列
  31. static PriorityBlockingQueue<Msg> pushQueue = new PriorityBlockingQueue<Msg>();
  32. static {
  33. //启动一个线程做真实推送
  34. new Thread(() -> {
  35. while (true) {
  36. Msg msg;
  37. try {
  38. long starTime = System.currentTimeMillis();
  39. //获取一条推送消息,此方法会进行阻塞,直到返回结果
  40. msg = pushQueue.take();
  41. //模拟推送耗时
  42. TimeUnit.MILLISECONDS.sleep(100);
  43. long endTime = System.currentTimeMillis();
  44. System.out.println(String.format("[%s,%s,take耗时:%s],%s,发送消息:%s", starTime, endTime, (endTime - starTime), Thread.currentThread().getName(), msg));
  45. } catch (InterruptedException e) {
  46. e.printStackTrace();
  47. }
  48. }
  49. }).start();
  50. }
  51. //推送消息,需要发送推送消息的调用该方法,会将推送信息先加入推送队列
  52. public static void pushMsg(int priority, String msg) throws InterruptedException {
  53. pushQueue.put(new Msg(priority, msg));
  54. }
  55. public static void main(String[] args) throws InterruptedException {
  56. for (int i = 5; i >= 1; i--) {
  57. String msg = "一起来学java高并发,第" + i + "天";
  58. Demo2.pushMsg(i, msg);
  59. }
  60. }
  61. }

输出:

  1. [1565598857028,1565598857129,take耗时:101],Thread-0,发送消息:Msg{priority=1, msg='一起来学java高并发,第1天'}
  2. [1565598857162,1565598857263,take耗时:101],Thread-0,发送消息:Msg{priority=2, msg='一起来学java高并发,第2天'}
  3. [1565598857263,1565598857363,take耗时:100],Thread-0,发送消息:Msg{priority=3, msg='一起来学java高并发,第3天'}
  4. [1565598857363,1565598857463,take耗时:100],Thread-0,发送消息:Msg{priority=4, msg='一起来学java高并发,第4天'}
  5. [1565598857463,1565598857563,take耗时:100],Thread-0,发送消息:Msg{priority=5, msg='一起来学java高并发,第5天'}

main中放入了5条推送信息,i作为消息的优先级按倒叙放入的,最终输出结果中按照优先级由小到大输出。注意Msg实现了Comparable接口,具有了比较功能。

同步阻塞队列,SynchronousQueue没有容量,与其他BlockingQueue不同,SynchronousQueue是一个不存储元素的BlockingQueue,每一个put操作必须要等待一个take操作,否则不能继续添加元素,反之亦然。SynchronousQueue 在现实中用的不多,线程池中有用到过,Executors.newCachedThreadPool()实现中用到了这个队列,当有任务丢入线程池的时候,如果已创建的工作线程都在忙于处理任务,则会新建一个线程来处理丢入队列的任务。

来个示例代码:

  1. package com.itsoku.chat25;
  2. import java.util.concurrent.PriorityBlockingQueue;
  3. import java.util.concurrent.SynchronousQueue;
  4. import java.util.concurrent.TimeUnit;
  5. /**
  6. * 跟着阿里p7学并发,微信公众号:javacode2018
  7. */
  8. public class Demo3 {
  9. static SynchronousQueue<String> queue = new SynchronousQueue<>();
  10. public static void main(String[] args) throws InterruptedException {
  11. new Thread(() -> {
  12. try {
  13. long starTime = System.currentTimeMillis();
  14. queue.put("java高并发系列,路人甲Java!");
  15. long endTime = System.currentTimeMillis();
  16. System.out.println(String.format("[%s,%s,take耗时:%s],%s", starTime, endTime, (endTime - starTime), Thread.currentThread().getName()));
  17. } catch (InterruptedException e) {
  18. e.printStackTrace();
  19. }
  20. }).start();
  21. //休眠5秒之后,从队列中take一个元素
  22. TimeUnit.SECONDS.sleep(5);
  23. System.out.println(System.currentTimeMillis() + "调用take获取并移除元素," + queue.take());
  24. }
  25. }

输出:

  1. 1565600421645调用take获取并移除元素,java高并发系列,路人甲Java!
  2. [1565600416645,1565600421645,take耗时:5000],Thread-0

main方法中启动了一个线程,调用queue.put方法向队列中丢入一条数据,调用的时候产生了阻塞,从输出结果中可以看出,直到take方法被调用时,put方法才从阻塞状态恢复正常。

DelayQueue是一个支持延时获取元素的无界阻塞队列,里面的元素全部都是“可延期”的元素,列头的元素是最先“到期”的元素,如果队列里面没有元素到期,是不能从列头获取元素的,哪怕有元素也不行,也就是说只有在延迟期到时才能够从队列中取元素。

需求:还是推送的业务,有时候我们希望早上9点或者其他指定的时间进行推送,如何实现呢?此时DelayQueue就派上用场了。

我们先看一下DelayQueue类的声明:

  1. public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
  2. implements BlockingQueue<E>

元素E需要实现接口Delayed,我们看一下这个接口的代码:

  1. public interface Delayed extends Comparable<Delayed> {
  2. long getDelay(TimeUnit unit);
  3. }

Delayed继承了Comparable接口,这个接口是用来做比较用的,DelayQueue内部使用PriorityQueue来存储数据的,PriorityQueue是一个优先级队列,丢入的数据会进行排序,排序方法调用的是Comparable接口中的方法。下面主要说一下Delayed接口中的getDelay方法:此方法在给定的时间单位内返回与此对象关联的剩余延迟时间。

对推送我们再做一下处理,让其支持定时发送(定时在将来某个时间也可以说是延迟发送),代码如下:

  1. package com.itsoku.chat25;
  2. import java.util.Calendar;
  3. import java.util.concurrent.DelayQueue;
  4. import java.util.concurrent.Delayed;
  5. import java.util.concurrent.PriorityBlockingQueue;
  6. import java.util.concurrent.TimeUnit;
  7. /**
  8. * 跟着阿里p7学并发,微信公众号:javacode2018
  9. */
  10. public class Demo4 {
  11. //推送信息封装
  12. static class Msg implements Delayed {
  13. //优先级,越小优先级越高
  14. private int priority;
  15. //推送的信息
  16. private String msg;
  17. //定时发送时间,毫秒格式
  18. private long sendTimeMs;
  19. public Msg(int priority, String msg, long sendTimeMs) {
  20. this.priority = priority;
  21. this.msg = msg;
  22. this.sendTimeMs = sendTimeMs;
  23. }
  24. @Override
  25. public String toString() {
  26. return "Msg{" +
  27. "priority=" + priority +
  28. ", msg='" + msg + '\'' +
  29. ", sendTimeMs=" + sendTimeMs +
  30. '}';
  31. }
  32. @Override
  33. public long getDelay(TimeUnit unit) {
  34. return unit.convert(this.sendTimeMs - Calendar.getInstance().getTimeInMillis(), TimeUnit.MILLISECONDS);
  35. }
  36. @Override
  37. public int compareTo(Delayed o) {
  38. if (o instanceof Msg) {
  39. Msg c2 = (Msg) o;
  40. return Integer.compare(this.priority, c2.priority);
  41. }
  42. return 0;
  43. }
  44. }
  45. //推送队列
  46. static DelayQueue<Msg> pushQueue = new DelayQueue<Msg>();
  47. static {
  48. //启动一个线程做真实推送
  49. new Thread(() -> {
  50. while (true) {
  51. Msg msg;
  52. try {
  53. //获取一条推送消息,此方法会进行阻塞,直到返回结果
  54. msg = pushQueue.take();
  55. //此处可以做真实推送
  56. long endTime = System.currentTimeMillis();
  57. System.out.println(String.format("定时发送时间:%s,实际发送时间:%s,发送消息:%s", msg.sendTimeMs, endTime, msg));
  58. } catch (InterruptedException e) {
  59. e.printStackTrace();
  60. }
  61. }
  62. }).start();
  63. }
  64. //推送消息,需要发送推送消息的调用该方法,会将推送信息先加入推送队列
  65. public static void pushMsg(int priority, String msg, long sendTimeMs) throws InterruptedException {
  66. pushQueue.put(new Msg(priority, msg, sendTimeMs));
  67. }
  68. public static void main(String[] args) throws InterruptedException {
  69. for (int i = 5; i >= 1; i--) {
  70. String msg = "一起来学java高并发,第" + i + "天";
  71. Demo4.pushMsg(i, msg, Calendar.getInstance().getTimeInMillis() + i * 2000);
  72. }
  73. }
  74. }

输出:

  1. 定时发送时间:1565603357198,实际发送时间:1565603357198,发送消息:Msg{priority=1, msg='一起来学java高并发,第1天', sendTimeMs=1565603357198}
  2. 定时发送时间:1565603359198,实际发送时间:1565603359198,发送消息:Msg{priority=2, msg='一起来学java高并发,第2天', sendTimeMs=1565603359198}
  3. 定时发送时间:1565603361198,实际发送时间:1565603361199,发送消息:Msg{priority=3, msg='一起来学java高并发,第3天', sendTimeMs=1565603361198}
  4. 定时发送时间:1565603363198,实际发送时间:1565603363199,发送消息:Msg{priority=4, msg='一起来学java高并发,第4天', sendTimeMs=1565603363198}
  5. 定时发送时间:1565603365182,实际发送时间:1565603365183,发送消息:Msg{priority=5, msg='一起来学java高并发,第5天', sendTimeMs=1565603365182}

可以看出时间发送时间,和定时发送时间基本一致,代码中Msg需要实现Delayed接口,重点在于getDelay方法,这个方法返回剩余的延迟时间,代码中使用this.sendTimeMs减去当前时间的毫秒格式时间,得到剩余延迟时间。

LinkedTransferQueue是一个由链表结构组成的无界阻塞TransferQueue队列。相对于其他阻塞队列,LinkedTransferQueue多了tryTransfer和transfer方法。

LinkedTransferQueue类继承自AbstractQueue抽象类,并且实现了TransferQueue接口:

  1. public interface TransferQueue<E> extends BlockingQueue<E> {
  2.     // 如果存在一个消费者已经等待接收它,则立即传送指定的元素,否则返回false,并且不进入队列。
  3.     boolean tryTransfer(E e);
  4.     // 如果存在一个消费者已经等待接收它,则立即传送指定的元素,否则等待直到元素被消费者接收。
  5.     void transfer(E e) throws InterruptedException;
  6.     // 在上述方法的基础上设置超时时间
  7.     boolean tryTransfer(E e, long timeout, TimeUnit unit)
  8.         throws InterruptedException;
  9.     // 如果至少有一位消费者在等待,则返回true
  10.     boolean hasWaitingConsumer();
  11.     // 获取所有等待获取元素的消费线程数量
  12.     int getWaitingConsumerCount();
  13. }

再看一下上面的这些方法,transfer(E e)方法和SynchronousQueue的put方法类似,都需要等待消费者取走元素,否者一直等待。其他方法和ArrayBlockingQueue、LinkedBlockingQueue中的方法类似。

  1. 重点需要了解BlockingQueue中的所有方法,以及他们的区别
  2. 重点掌握ArrayBlockingQueueLinkedBlockingQueuePriorityBlockingQueueDelayQueue的使用场景
  3. 需要处理的任务有优先级的,使用PriorityBlockingQueue
  4. 处理的任务需要延时处理的,使用DelayQueue
  1. 第1天:必须知道的几个概念
  2. 第2天:并发级别
  3. 第3天:有关并行的两个重要定律
  4. 第4天:JMM相关的一些概念
  5. 第5天:深入理解进程和线程
  6. 第6天:线程的基本操作
  7. 第7天:volatile与Java内存模型
  8. 第8天:线程组
  9. 第9天:用户线程和守护线程
  10. 第10天:线程安全和synchronized关键字
  11. 第11天:线程中断的几种方式
  12. 第12天JUC:ReentrantLock重入锁
  13. 第13天:JUC中的Condition对象
  14. 第14天:JUC中的LockSupport工具类,必备技能
  15. 第15天:JUC中的Semaphore(信号量)
  16. 第16天:JUC中等待多线程完成的工具类CountDownLatch,必备技能
  17. 第17天:JUC中的循环栅栏CyclicBarrier的6种使用场景
  18. 第18天:JAVA线程池,这一篇就够了
  19. 第19天:JUC中的Executor框架详解1
  20. 第20天:JUC中的Executor框架详解2
  21. 第21天:java中的CAS,你需要知道的东西
  22. 第22天:JUC底层工具类Unsafe,高手必须要了解
  23. 第23天:JUC中原子类,一篇就够了
  24. 第24天:ThreadLocal、InheritableThreadLocal(通俗易懂)

java高并发系列连载中,总计估计会有四五十篇文章。

阿里p7一起学并发,公众号:路人甲java,每天获取最新文章!

版权声明:本文为itsoku123原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://www.cnblogs.com/itsoku123/p/11358757.html