34.Java 阻塞队列（阻塞队列架构、阻塞队列分类、阻塞队列核心方法）

一、阻塞队列概述

• java.util.concurrent 包下的 BlockingQueue 接口很好的解决了多线程中如何高效安全传输数据的问题，可以使用这些高效并且线程安全的队列类快速搭建高质量的多线程程序

• 阻塞队列通过一个共享的队列，使得数据由队列的一端输入，从另外一端输出，当队列空时，从队列中获取元素的操作将会被阻塞，当队列满时，从队列中添加元素的操作将会被阻塞

• 试图从空的队列中获取元素的线程将会被阻塞，直到其他线程往空的队列插入新的元素，试图向已满的队列中添加新元素的线程将会被阻塞，直到其他线程从队列中移除一个或多个元素或者完全清空

• 所谓阻塞，即在某些情况下将线程挂起，一旦条件满足，被挂起的线程会自动被唤起

• 多线程环境中，通过队列可以很容易实现数据共享，比如经典的生产者和消费者模型，通过队列可以很便利地实现两者之间的数据共享

• 当队列中没有数据时，消费者端的所有线程都会被阻塞，直到有数据存入队列，当队列中填满数据的情况下，生产者端的所有线程都会被阻塞，直到队列中有空间

二、阻塞队列架构

• BlockingQueue 接口的超接口有 Collection、Iterable、Queue

• BlockingQueue 接口的实现类有 ArrayBlockingQueue、DelayQueue、LinkedBlockingDeque、LinkedBlockingQueue、LinkedTransferQueue、PriorityBlockingQueue、SynchronousQueue

三、阻塞队列分类

1、ArrayBlockingQueue 类

• ArrayBlockingQueue 是由数组结构实现的有界阻塞队列

• ArrayBlockingQueue 在生产者存放数据和消费者获取数据时都是共用同一个锁对象，无法并行

2、LinkedBlockingQueue 类

• LinkedBlockingQueue 是由链表结构实现的有界阻塞队列，大小默认值为 Integer.MAX_VALUE

• LinkedBlockingQueue 对于生产者端和消费者端分别采用了独立的锁来控制数据同步，在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能

3、DelayQueue 类

• 基于优先级队列实现的延迟无界阻塞队列

• DelayQueue 中的数据只有当其达到指定的延迟时间，才能够从队列中获取到该数据

• DelayQueue 是一个没有大小限制的队列，因此往队列中插入数据的操作（生产者）不会被阻塞，只有获取数据的操作（消费者）才会被阻塞

4、PriorityBlockingQueue 类

• PriorityBlockingQueue 是基于优先级排序实现的无界阻塞队列

• PriorityBlockingQueue 不会阻塞数据生产者，只会在没有可消费的数据时，阻塞数据的消费者

5、SynchronousQueue 类

• SynchronousQueue 是无缓冲的等待队列

• 相对于有缓冲的 BlockingQueue 来说，SynchronousQueue 少了一个中间经销商的环节（缓冲区）

• SynchronousQueue 是不存储元素的阻塞队列，也即单个元素的队列

• 声明一个 SynchronousQueue 对象有公平模式和非公平模式两种方式

（1）公平模式

• SynchronousQueue 使用公平锁，并配合一个先进先出（FIFO）队列来阻塞多余的生产者和消费者，从而体系整体的公平策略

（2）非公平模式

• SynchronousQueue 使用非公平锁，同时配合一个后进先出（LIFO）队列来管理多余的生产者和消费者

• 这种模式，如果生产者和消费者的数据处理速度有差距，则很容易出现线程饥饿饥饿的情况，即可能有某些生产者或者是消费者的数据永远都得不到处理

6、LinkedTransferQueue 类

• LinkedTransferQueue 是基于链表结构实现的无界阻塞队列

• LinkedTransferQueue 采用一种预占模式，当消费者线程取数据时，如果队列不为空，则直接取走数据，若队列为空，那就生成一个节点数据为 null 的节点入队，然后消费者线程在该节点上等待，之后生产者线程入队时发现有一个节点数据为 null 的节点，生产者线程就不入对了，直接将数据填充到该节点，并唤醒该节点等待的消费者线程，被唤醒的消费者线程取走数据，从调用的方法返回

7、LinkedBlockingDeque 类

• LinkedBlockingDeque 是基于链表结构组成的双向阻塞队列

• LinkedBlockingDeque 有两种阻塞情况

（1）插入数据时阻塞

• 当队列满时会进入阻塞状态，直到队列有空的位置时才可插入数据

• 插入数据的操作可以通过设置超时时间，超时后返回 false 表示操作失败，也可以不设置超时时间，但是会被一直阻塞，被中断后会抛出 InterruptedException 异常

（2）读取数据时阻塞

• 当队列空时会进入阻塞状态，直到队列不为空才可读取数据

• 读取数据的操作同样可以设置超时时间

四、阻塞队列核心方法

• BlockingQueue 的核心方法可以根据队列已满或为空时的执行情况分为抛出异常组、返回特殊值组、阻塞组、超时组

1、抛出异常组

（1）基本介绍

（2）演示

• add 方法演示

BlockingQueue blockingQueue = new ArrayBlockingQueue(3);

System.out.println(blockingQueue.add(1)); // true
System.out.println(blockingQueue.add(2)); // true
System.out.println(blockingQueue.add(3)); // true
System.out.println(blockingQueue.add(4)); // 抛出 java.lang.IllegalStateException 异常

• remove 方法演示

BlockingQueue blockingQueue = new ArrayBlockingQueue(3);

System.out.println(blockingQueue.add(1)); // true
System.out.println(blockingQueue.add(2)); // true
System.out.println(blockingQueue.add(3)); // true

System.out.println(blockingQueue.remove()); // 1
System.out.println(blockingQueue.remove()); // 2
System.out.println(blockingQueue.remove()); // 3
System.out.println(blockingQueue.remove()); // 抛出 java.util.NoSuchElementException 异常

• element 方法演示

BlockingQueue blockingQueue = new ArrayBlockingQueue(3);

System.out.println(blockingQueue.add(1)); // true
System.out.println(blockingQueue.add(2)); // true
System.out.println(blockingQueue.add(3)); // true

System.out.println(blockingQueue.remove()); // 1
System.out.println(blockingQueue.element()); // 2

System.out.println(blockingQueue.remove()); // 2
System.out.println(blockingQueue.element()); // 3

System.out.println(blockingQueue.element()); // 抛出 java.util.NoSuchElementException 异常

2、返回特殊值组

（1）基本介绍

（2）演示

• offer 方法演示

BlockingQueue blockingQueue = new ArrayBlockingQueue(3);

System.out.println(blockingQueue.offer(1)); // true
System.out.println(blockingQueue.offer(2)); // true
System.out.println(blockingQueue.offer(3)); // true
System.out.println(blockingQueue.offer(4)); // false

• poll 方法演示

BlockingQueue blockingQueue = new ArrayBlockingQueue(3);

System.out.println(blockingQueue.offer(1)); // true
System.out.println(blockingQueue.offer(2)); // true
System.out.println(blockingQueue.offer(3)); // true

System.out.println(blockingQueue.poll()); // 1
System.out.println(blockingQueue.poll()); // 2
System.out.println(blockingQueue.poll()); // 3
System.out.println(blockingQueue.poll()); // null

• peek 方法演示

BlockingQueue blockingQueue = new ArrayBlockingQueue(3);

System.out.println(blockingQueue.offer(1)); // true
System.out.println(blockingQueue.offer(2)); // true
System.out.println(blockingQueue.offer(3)); // true

System.out.println(blockingQueue.poll()); // 1
System.out.println(blockingQueue.peek()); // 2

System.out.println(blockingQueue.poll()); // 2
System.out.println(blockingQueue.peek()); // 3

System.out.println(blockingQueue.poll()); // 3
System.out.println(blockingQueue.peek()); // null

3、阻塞组

（1）基本介绍

（2）演示

• put 方法演示

BlockingQueue blockingQueue = new ArrayBlockingQueue(3);

blockingQueue.put(1);
System.out.println("a"); // a
        
blockingQueue.put(2);
System.out.println("b"); // b
        
blockingQueue.put(3);
System.out.println("c"); // c
        
blockingQueue.put(4); // 阻塞
System.out.println("d"); // 阻塞

• take 方法

BlockingQueue blockingQueue = new ArrayBlockingQueue(3);

blockingQueue.put(1);
blockingQueue.put(2);
blockingQueue.put(3);

System.out.println(blockingQueue.take()); // 1
System.out.println(blockingQueue.take()); // 2
System.out.println(blockingQueue.take()); // 3
System.out.println(blockingQueue.take()); // 阻塞

4、超时组

（1）基本介绍

（2）演示

• offer 方法演示

BlockingQueue blockingQueue = new ArrayBlockingQueue(3);

System.out.println(blockingQueue.offer(1, 3L, TimeUnit.SECONDS)); // true
System.out.println(blockingQueue.offer(2, 3L, TimeUnit.SECONDS)); // true
System.out.println(blockingQueue.offer(3, 3L, TimeUnit.SECONDS)); // true
System.out.println(blockingQueue.offer(4, 3L, TimeUnit.SECONDS)); // 超时等待，超时等待后返回 false

• poll 方法演示

BlockingQueue blockingQueue = new ArrayBlockingQueue(3);

System.out.println(blockingQueue.offer(1, 3L, TimeUnit.SECONDS)); // true
System.out.println(blockingQueue.offer(2, 3L, TimeUnit.SECONDS)); // true
System.out.println(blockingQueue.offer(3, 3L, TimeUnit.SECONDS)); // true

System.out.println(blockingQueue.poll(3L, TimeUnit.SECONDS)); // 1
System.out.println(blockingQueue.poll(3L, TimeUnit.SECONDS)); // 2
System.out.println(blockingQueue.poll(3L, TimeUnit.SECONDS)); // 3
System.out.println(blockingQueue.poll(3L, TimeUnit.SECONDS)); // 超时等待，超时等待后返回 false