每天早上七点三十，准时推送干货 一、摘要在之前的文章中，我们介绍了生产者和消费者模型的最基本实现思路，相信大家对它已经有一个初步的认识。 在 Java 的并发包里面还有一个非常重要的接口：BlockingQueue。 BlockingQueue是一个阻塞队列，更为准确的解释是：BlockingQueue是一个基于阻塞机制实现的线程安全的队列。通过它也可以实现生产者和消费者模型，并且效率更高、安全可靠，相比之前介绍的生产者和消费者模型，它可以同时实现生产者和消费者并行运行。 那什么是阻塞队列呢？ 简单的说，就是当参数在入队和出队时，通过加锁的方式来避免线程并发操作时导致的数据异常问题。 在 Java 中，能对线程并发执行进行加锁的方式主要有synchronized和ReentrantLock，其中BlockingQueue采用的是ReentrantLock方式实现。 与此对应的还有非阻塞机制的队列，主要是采用 CAS 方式来控制并发操作，例如：ConcurrentLinkedQueue，这个我们在后面的文章再进行分享介绍。 今天我们主要介绍BlockingQueue相关的知识和用法，废话不多说了，进入正题！ 二、BlockingQueue 方法介绍打开BlockingQueue的源码，你会发现它继承自Queue，正如上文提到的，它本质是一个队列接口。 代码语言：javascript代码运行次数：0运行复制public interface BlockingQueue //...省略 } 关于队列，我们在之前的集合系列文章中对此有过深入的介绍，本篇就再次简单的介绍一下。 队列其实是一个数据结构，元素遵循先进先出的原则，所有新元素的插入，也被称为入队操作，会插入到队列的尾部；元素的移除，也被称为出队操作，会从队列的头部开始移除，从而保证先进先出的原则。 在Queue接口中，总共有 6 个方法，可以分为 3 类，分别是：插入、移除、查询，内容如下： 方法 描述 add(e) 插入元素，如果插入失败，就抛异常 offer(e) 插入元素，如果插入成功，就返回 true；反之 false remove() 移除元素，如果移除失败，就抛异常 poll() 移除元素，如果移除成功，返回 true；反之 false element() 获取队首元素，如果获取结果为空，就抛异常 peek() 获取队首元素，如果获取结果为空，返回空对象 因为BlockingQueue是Queue的子接口，了解Queue接口里面的方法，有助于我们对BlockingQueue的理解。 除此之外，BlockingQueue还单独扩展了一些特有的方法，内容如下： 方法 描述 put(e) 插入元素，如果没有插入成功，线程会一直阻塞，直到队列中有空间再继续 offer(e, time, unit) 插入元素，如果在指定的时间内没有插入成功，就返回 false；反之 true take() 移除元素，如果没有移除成功，线程会一直阻塞，直到队列中新的数据被加入 poll(time, unit) 移除元素，如果在指定的时间内没有移除成功，就返回 false；反之 true drainTo(Collection c, int maxElements) 一次性取走队列中的数据到 c 中，可以指定取的个数。该方法可以提升获取数据效率，不需要多次分批加锁或释放锁 分析源码，你会发现相比普通的Queue子类，BlockingQueue子类主要有以下几个明显的不同点： 1.元素插入和移除时线程安全：主要是通过在入队和出队时进行加锁，保证了队列线程安全，加锁逻辑采用ReentrantLock实现2.支持阻塞的入队和出队方法：当队列满时，会阻塞入队的线程，直到队列不满；当队列为空时，会阻塞出队的线程，直到队列中有元素；同时支持超时机制，防止线程一直阻塞三、BlockingQueue 用法详解打开源码，BlockingQueue接口的实现类非常多，我们重点讲解一下其中的 5 个非常重要的实现类，分别如下表所示。 实现类 功能 ArrayBlockingQueue 基于数组的阻塞队列，使用数组存储数据，需要指定长度，所以是一个有界队列 LinkedBlockingQueue 基于链表的阻塞队列，使用链表存储数据，默认是一个无界队列；也可以通过构造方法中的capacity设置最大元素数量，所以也可以作为有界队列 SynchronousQueue 一种没有缓冲的队列，生产者产生的数据直接会被消费者获取并且立刻消费 PriorityBlockingQueue 基于优先级别的阻塞队列，底层基于数组实现，是一个无界队列 DelayQueue 延迟队列，其中的元素只有到了其指定的延迟时间，才能够从队列中出队 下面我们对以上实现类的用法，进行一一介绍。 3.1、ArrayBlockingQueueArrayBlockingQueue是一个基于数组的阻塞队列，初始化的时候必须指定队列大小，源码实现比较简单，采用的是ReentrantLock和Condition实现生产者和消费者模型，部分核心源码如下： 代码语言：javascript代码运行次数：0运行复制public class ArrayBlockingQueue implements BlockingQueue /** 使用数组存储队列中的元素 */ final Object[] items; /** 使用独占锁ReetrantLock */ final ReentrantLock lock; /** 等待出队的条件 */ private final Condition notEmpty; /** 等待入队的条件 */ private final Condition notFull; /** 初始化时，需要指定队列大小 */ public ArrayBlockingQueue(int capacity) { this(capacity, false); } /** 初始化时，也指出指定是否为公平锁， */ public ArrayBlockingQueue(int capacity, boolean fair) { if (capacity <= 0) throw new IllegalArgumentException(); this.items = new Object[capacity]; lock = new ReentrantLock(fair); notEmpty = lock.newCondition(); notFull = lock.newCondition(); } /**入队操作*/ public void put(E e) throws InterruptedException { checkNotNull(e); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == items.length) notFull.await(); enqueue(e); } finally { lock.unlock(); } } /**出队操作*/ public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == 0) notEmpty.await(); return dequeue(); } finally { lock.unlock(); } } } ArrayBlockingQueue采用ReentrantLock进行加锁，只有一个ReentrantLock对象，这意味着生产者和消费者无法并行运行。 我们看一个简单的示例代码如下： 代码语言：javascript代码运行次数：0运行复制public class Container { /** * 初始化阻塞队列 */ private final BlockingQueue /** * 添加数据到阻塞队列 * @param value */ public void add(Integer value) { try { queue.put(value); System.out.println("生产者："+ Thread.currentThread().getName()+"，add：" + value); } catch (InterruptedException e) { e.printStackTrace(); } } /** * 从阻塞队列获取数据 */ public void get() { try { Integer value = queue.take(); System.out.println("消费者："+ Thread.currentThread().getName()+"，value：" + value); } catch (InterruptedException e) { e.printStackTrace(); } } } 代码语言：javascript代码运行次数：0运行复制/** * 生产者 */ public class Producer extends Thread { private Container container; public Producer(Container container) { this.container = container; } @Override public void run() { for (int i = 0; i < 6; i++) { container.add(i); } } } 代码语言：javascript代码运行次数：0运行复制/** * 消费者 */ public class Consumer extends Thread { private Container container; public Consumer(Container container) { this.container = container; } @Override public void run() { for (int i = 0; i < 6; i++) { container.get(); } } } 代码语言：javascript代码运行次数：0运行复制/** * 测试类 */ public class MyThreadTest { public static void main(String[] args) { Container container = new Container(); Producer producer = new Producer(container); Consumer consumer = new Consumer(container); producer.start(); consumer.start(); } } 运行结果如下： 代码语言：javascript代码运行次数：0运行复制生产者：Thread-0，add：0 生产者：Thread-0，add：1 生产者：Thread-0，add：2 生产者：Thread-0，add：3 生产者：Thread-0，add：4 生产者：Thread-0，add：5 消费者：Thread-1，value：0 消费者：Thread-1，value：1 消费者：Thread-1，value：2 消费者：Thread-1，value：3 消费者：Thread-1，value：4 消费者：Thread-1，value：5 可以很清晰的看到，生产者线程执行完毕之后，消费者线程才开始消费。 3.2、LinkedBlockingQueueLinkedBlockingQueue是一个基于链表的阻塞队列，初始化的时候无须指定队列大小，默认队列长度为Integer.MAX_VALUE，也就是 int 型最大值。 同样的，采用的是ReentrantLock和Condition实现生产者和消费者模型，不同的是它使用了两个lock，这意味着生产者和消费者可以并行运行，程序执行效率进一步得到提升。 部分核心源码如下： 代码语言：javascript代码运行次数：0运行复制public class LinkedBlockingQueue implements BlockingQueue /** 使用出队独占锁ReetrantLock */ private final ReentrantLock takeLock = new ReentrantLock(); /** 等待出队的条件 */ private final Condition notEmpty = takeLock.newCondition(); /** 使用入队独占锁ReetrantLock */ private final ReentrantLock putLock = new ReentrantLock(); /** 等待入队的条件 */ private final Condition notFull = putLock.newCondition(); /**入队操作*/ public void put(E e) throws InterruptedException { if (e == null) throw new NullPointerException(); int c = -1; Node final ReentrantLock putLock = this.putLock; final AtomicInteger count = this.count; putLock.lockInterruptibly(); try { while (count.get() == capacity) { notFull.await(); } enqueue(node); c = count.getAndIncrement(); if (c + 1 < capacity) notFull.signal(); } finally { putLock.unlock(); } if (c == 0) signalNotEmpty(); } /**出队操作*/ public E take() throws InterruptedException { E x; int c = -1; final AtomicInteger count = this.count; final ReentrantLock takeLock = this.takeLock; takeLock.lockInterruptibly(); try { while (count.get() == 0) { notEmpty.await(); } x = dequeue(); c = count.getAndDecrement(); if (c > 1) notEmpty.signal(); } finally { takeLock.unlock(); } if (c == capacity) signalNotFull(); return x; } } 把最上面的样例Container中的阻塞队列实现类换成LinkedBlockingQueue，调整如下： 代码语言：javascript代码运行次数：0运行复制/** * 初始化阻塞队列 */ private final BlockingQueue 再次运行结果如下： 代码语言：javascript代码运行次数：0运行复制生产者：Thread-0，add：0 消费者：Thread-1，value：0 生产者：Thread-0，add：1 消费者：Thread-1，value：1 生产者：Thread-0，add：2 消费者：Thread-1，value：2 生产者：Thread-0，add：3 生产者：Thread-0，add：4 生产者：Thread-0，add：5 消费者：Thread-1，value：3 消费者：Thread-1，value：4 消费者：Thread-1，value：5 可以很清晰的看到，生产者线程和消费者线程，交替并行执行。 3.3、SynchronousQueueSynchronousQueue是一个没有缓冲的队列，生产者产生的数据直接会被消费者获取并且立刻消费，相当于传统的一个请求对应一个应答模式。 相比ArrayBlockingQueue和LinkedBlockingQueue，SynchronousQueue实现机制也不同，它主要采用队列和栈来实现数据的传递，中间不存储任何数据，生产的数据必须得消费者处理，线程阻塞方式采用 JDK 提供的LockSupport park/unpark函数来完成，也支持公平和非公平两种模式。 当采用公平模式时：使用一个 FIFO 队列来管理多余的生产者和消费者当采用非公平模式时：使用一个 LIFO 栈来管理多余的生产者和消费者，这也是SynchronousQueue默认的模式部分核心源码如下： 代码语言：javascript代码运行次数：0运行复制public class SynchronousQueue implements BlockingQueue /**不同的策略实现*/ private transient volatile Transferer /**默认非公平模式*/ public SynchronousQueue() { this(false); } /**可以选策略，也可以采用公平模式*/ public SynchronousQueue(boolean fair) { transferer = fair ? new TransferQueue } /**入队操作*/ public void put(E e) throws InterruptedException { if (e == null) throw new NullPointerException(); if (transferer.transfer(e, false, 0) == null) { Thread.interrupted(); throw new InterruptedException(); } } /**出队操作*/ public E take() throws InterruptedException { E e = transferer.transfer(null, false, 0); if (e != null) return e; Thread.interrupted(); throw new InterruptedException(); } } 同样的，把最上面的样例Container中的阻塞队列实现类换成SynchronousQueue，代码如下： 代码语言：javascript代码运行次数：0运行复制public class Container { /** * 初始化阻塞队列 */ private final BlockingQueue /** * 添加数据到阻塞队列 * @param value */ public void add(Integer value) { try { queue.put(value); Thread.sleep(100); System.out.println("生产者："+ Thread.currentThread().getName()+"，add：" + value); } catch (InterruptedException e) { e.printStackTrace(); } } /** * 从阻塞队列获取数据 */ public void get() { try { Integer value = queue.take(); Thread.sleep(200); System.out.println("消费者："+ Thread.currentThread().getName()+"，value：" + value); } catch (InterruptedException e) { e.printStackTrace(); } } } 再次运行结果如下： 代码语言：javascript代码运行次数：0运行复制生产者：Thread-0，add：0 消费者：Thread-1，value：0 生产者：Thread-0，add：1 消费者：Thread-1，value：1 生产者：Thread-0，add：2 消费者：Thread-1，value：2 生产者：Thread-0，add：3 消费者：Thread-1，value：3 生产者：Thread-0，add：4 消费者：Thread-1，value：4 生产者：Thread-0，add：5 消费者：Thread-1，value：5 可以很清晰的看到，生产者线程和消费者线程，交替串行执行，生产者每投递一条数据，消费者处理一条数据。 3.4、PriorityBlockingQueuePriorityBlockingQueue是一个基于优先级别的阻塞队列，底层基于数组实现，可以认为是一个无界队列。 PriorityBlockingQueue与ArrayBlockingQueue的实现逻辑，基本相似，也是采用ReentrantLock来实现加锁的操作。 最大不同点在于： 1.PriorityBlockingQueue内部基于数组实现的最小二叉堆算法，可以对队列中的元素进行排序，插入队列的元素需要实现Comparator或者Comparable接口，以便对元素进行排序2.其次，队列的长度是可扩展的，不需要显式指定长度，上限为Integer.MAX_VALUE - 8部分核心源码如下： 代码语言：javascript代码运行次数：0运行复制public class PriorityBlockingQueue implements BlockingQueue /**队列元素*/ private transient Object[] queue; /**比较器*/ private transient Comparator comparator; /**采用ReentrantLock进行加锁*/ private final ReentrantLock lock; /**条件等待与通知*/ private final Condition notEmpty; /**入队操作*/ public boolean offer(E e) { if (e == null) throw new NullPointerException(); final ReentrantLock lock = this.lock; lock.lock(); int n, cap; Object[] array; while ((n = size) >= (cap = (array = queue).length)) tryGrow(array, cap); try { Comparator cmp = comparator; if (cmp == null) siftUpComparable(n, e, array); else siftUpUsingComparator(n, e, array, cmp); size = n + 1; notEmpty.signal(); } finally { lock.unlock(); } return true; } /**出队操作*/ public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); E result; try { while ( (result = dequeue()) == null) notEmpty.await(); } finally { lock.unlock(); } return result; } } 同样的，把最上面的样例Container中的阻塞队列实现类换成PriorityBlockingQueue，调整如下： 代码语言：javascript代码运行次数：0运行复制/** * 初始化阻塞队列 */ private final BlockingQueue 生产者插入数据的内容，我们改下插入顺序。 代码语言：javascript代码运行次数：0运行复制/** * 生产者 */ public class Producer extends Thread { private Container container; public Producer(Container container) { this.container = container; } @Override public void run() { container.add(5); container.add(3); container.add(1); container.add(2); container.add(0); container.add(4); } } 最后运行结果如下： 代码语言：javascript代码运行次数：0运行复制生产者：Thread-0，add：5 生产者：Thread-0，add：3 生产者：Thread-0，add：1 生产者：Thread-0，add：2 生产者：Thread-0，add：0 生产者：Thread-0，add：4 消费者：Thread-1，value：0 消费者：Thread-1，value：1 消费者：Thread-1，value：2 消费者：Thread-1，value：3 消费者：Thread-1，value：4 消费者：Thread-1，value：5 从日志上可以很明显看出，对于整数，默认情况下，按照升序排序，消费者默认从 0 开始处理。 3.5、DelayQueueDelayQueue是一个线程安全的延迟队列，存入队列的元素不会立刻被消费，只有到了其指定的延迟时间，才能够从队列中出队。 底层采用的是PriorityQueue来存储元素，DelayQueue的特点在于：插入队列中的数据可以按照自定义的delay时间进行排序，快到期的元素会排列在前面，只有delay时间小于 0 的元素才能够被取出。 部分核心源码如下： 代码语言：javascript代码运行次数：0运行复制public class DelayQueue implements BlockingQueue /**采用ReentrantLock进行加锁*/ private final transient ReentrantLock lock = new ReentrantLock(); /**采用PriorityQueue进行存储数据*/ private final PriorityQueue /**条件等待与通知*/ private final Condition available = lock.newCondition(); /**入队操作*/ public boolean offer(E e) { final ReentrantLock lock = this.lock; lock.lock(); try { q.offer(e); if (q.peek() == e) { leader = null; available.signal(); } return true; } finally { lock.unlock(); } } /**出队操作*/ public E poll() { final ReentrantLock lock = this.lock; lock.lock(); try { E first = q.peek(); if (first == null || first.getDelay(NANOSECONDS) > 0) return null; else return q.poll(); } finally { lock.unlock(); } } } 同样的，把最上面的样例Container中的阻塞队列实现类换成DelayQueue，代码如下： 代码语言：javascript代码运行次数：0运行复制public class Container { /** * 初始化阻塞队列 */ private final BlockingQueue /** * 添加数据到阻塞队列 * @param value */ public void add(DelayedUser value) { try { queue.put(value); System.out.println("生产者："+ Thread.currentThread().getName()+"，add：" + value); } catch (InterruptedException e) { e.printStackTrace(); } } /** * 从阻塞队列获取数据 */ public void get() { try { DelayedUser value = queue.take(); String time = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()); System.out.println(time + " 消费者："+ Thread.currentThread().getName()+"，value：" + value); } catch (InterruptedException e) { e.printStackTrace(); } } } DelayQueue队列中的元素需要显式实现Delayed接口，定义一个DelayedUser类，代码如下： 代码语言：javascript代码运行次数：0运行复制public class DelayedUser implements Delayed { /** * 当前时间戳 */ private long start; /** * 延迟时间(单位：毫秒) */ private long delayedTime; /** * 名称 */ private String name; public DelayedUser(long delayedTime, String name) { this.start = System.currentTimeMillis(); this.delayedTime = delayedTime; this.name = name; } @Override public long getDelay(TimeUnit unit) { // 获取当前延迟的时间 long diffTime = (start + delayedTime) - System.currentTimeMillis(); return unit.convert(diffTime,TimeUnit.MILLISECONDS); } @Override public int compareTo(Delayed o) { // 判断当前对象的延迟时间是否大于目标对象的延迟时间 return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS)); } @Override public String toString() { return "DelayedUser{" + "delayedTime=" + delayedTime + ", name='" + name + '\'' + '}'; } } 生产者插入数据的内容，做如下调整。 代码语言：javascript代码运行次数：0运行复制/** * 生产者 */ public class Producer extends Thread { private Container container; public Producer(Container container) { this.container = container; } @Override public void run() { for (int i = 0; i < 6; i++) { container.add(new DelayedUser(1000 * i, "张三" + i)); } } } 最后运行结果如下： 代码语言：javascript代码运行次数：0运行复制生产者：Thread-0，add：DelayedUser{delayedTime=0, name='张三0'} 生产者：Thread-0，add：DelayedUser{delayedTime=1000, name='张三1'} 生产者：Thread-0，add：DelayedUser{delayedTime=2000, name='张三2'} 生产者：Thread-0，add：DelayedUser{delayedTime=3000, name='张三3'} 生产者：Thread-0，add：DelayedUser{delayedTime=4000, name='张三4'} 生产者：Thread-0，add：DelayedUser{delayedTime=5000, name='张三5'} 2023-11-03 14:55:33 消费者：Thread-1，value：DelayedUser{delayedTime=0, name='张三0'} 2023-11-03 14:55:34 消费者：Thread-1，value：DelayedUser{delayedTime=1000, name='张三1'} 2023-11-03 14:55:35 消费者：Thread-1，value：DelayedUser{delayedTime=2000, name='张三2'} 2023-11-03 14:55:36 消费者：Thread-1，value：DelayedUser{delayedTime=3000, name='张三3'} 2023-11-03 14:55:37 消费者：Thread-1，value：DelayedUser{delayedTime=4000, name='张三4'} 2023-11-03 14:55:38 消费者：Thread-1，value：DelayedUser{delayedTime=5000, name='张三5'} 可以很清晰的看到，延迟时间最低的排在最前面。 四、小结最后我们来总结一下BlockingQueue阻塞队列接口，它提供了很多非常丰富的生产者和消费者模型的编程实现，同时兼顾了线程安全和执行效率的特点。 开发者可以通过BlockingQueue阻塞队列接口，简单的代码编程即可实现多线程中数据高效安全传输的目的，确切的说，它帮助开发者减轻了不少的编程难度。 在实际的业务开发中，其中LinkedBlockingQueue使用的是最广泛的，因为它的执行效率最高，在使用的时候，需要平衡好队列长度，防止过大导致内存溢出。 举个最简单的例子，比如某个功能上线之后，需要做下压力测试，总共需要请求 10000 次，采用 100 个线程去执行，测试服务是否能正常工作。如何实现呢？ 可能有的同学想到，每个线程执行 100 次请求，启动 100 个线程去执行，可以是可以，就是有点笨拙。 其实还有另一个办法，就是将 10000 个请求对象，存入到阻塞队列中，然后采用 100 个线程去消费执行，这种编程模型会更佳灵活。 具体示例代码如下： 代码语言：javascript代码运行次数：0运行复制public static void main(String[] args) throws InterruptedException { // 将每个用户访问百度服务的请求任务，存入阻塞队列中 // 也可以也采用多线程写入 BlockingQueue for (int i = 0; i < 10000; i++) { queue.put("https://www.baidu.com?paramKey=" + i); } // 模拟100个线程，执行10000次请求访问百度 final int threadNum = 100; for (int i = 0; i < threadNum; i++) { final int threadCount = i + 1; new Thread(new Runnable() { @Override public void run() { System.out.println("thread " + threadCount + " start"); boolean over = false; while (!over) { String url = queue.poll(); if(Objects.nonNull(url)) { // 发起请求 String result =HttpUtils.getUrl(url); System.out.println("thread " + threadCount + " run result:" + result); }else { // 任务结束 over = true; System.out.println("thread " + threadCount + " final"); } } } }).start(); } } 本文主要围绕BlockingQueue阻塞队列接口，从方法介绍到用法详解，做了一次知识总结，如果有描述不对的地方，欢迎留言指出！ 五、参考1. https://www.cnblogs.com/xrq730/p/4855857.html 2. https://juejin.cn/post/6999798721269465102 喜欢就分享 认同就点赞 支持就在看 一键四连，你的offer也四连