【JUC】八、阻塞队列

news2024/11/25 2:56:23

文章目录

1、阻塞队列概述
2、阻塞队列分类
3、阻塞队列的四组核心方法
4、Demo

队列，先进先出，类似排队
栈，先进后出，用于要优先处理最近发生的事件的场景

1、阻塞队列概述

阻塞队列，一个生产消费模式，当：

队列放满了，put不进去了，put的线程阻塞（挂起），等可以put了，再唤起
取没了，只能等待，取的线程阻塞（挂起），等可以取了，再唤起

而亮点就在于什么时候阻塞线程，什么时候唤起线程，则由BlockingQueue一手包办。

在这里插入图片描述

2、阻塞队列分类

在这里插入图片描述

ArrayBlockingQueue：底层是一个定长数组，有界阻塞队列
LinkedBlockingQueue：底层是一个链表，有界阻塞队列，但它的默认值其实足够大了（大小默认值为Integer.MAX_VALUE）
DelayQueue：队列中的元素只有到了指定的延迟时间，才能获取到该元素。是无界队列，因此生产者线程永不阻塞
PriorityBlockingQueue：支持优先级排序的无界阻塞队列（优先级的判断通过其构造方法传入Compator对象决定）
SynchronousQueue：无中介，一种不存储元素的阻塞队列、单个元素的队列，一个线程写入了数据，就必须得有一个线程取，否则不能再继续添加，用于传递性的场景
LinkedTransferQueue：底层是一个链表，无界阻塞队列。亮点是其有预占模式，其消费者线程取元素时，若队列为空，就生成一个元素为null的节点入队，消费者线程就等待在这个节点上，后续生产者线程来了发现有个元素为null的节点，就不再入队，直接把数据填充给这个null节点，并唤醒改null节点上等待的消费者线程取走元素
LinkedBlockingDeque：底层是一个链表，双向阻塞队列，可以从队列的两端插入和移除元素

3、阻塞队列的四组核心方法

插入相关的有：add、offer、put
移除相关的有：remove、poll、take
检查相关的有：element、peek

按照队列空或者队列满时的表现，可分为以下四组：

在这里插入图片描述

以列为单位来看以上表格的含义：

抛出异常列，插入、移除、检查对应三个方法：add、remove、element，这一组，是队列满或空时再调就抛出异常
特殊值列，插入、移除、检查对应三个方法：offer、poll、peek，这一组，是队列满时调offer插入返回false，而不是抛异常
阻塞列，插入、移除对应两个方法：put、take，这一组，是队列满时再调put就一直阻塞，直到put成功，take同理
超时列，插入、移除对应两个方法：offer、poll，这一组，是队列满时，再调offer来放数据，会阻塞，但有个最大时间，超过这个时间生产者线程就自动退出

//创建阻塞队列
BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<>(3);
//写
for (int i = 0; i < 4; i++) {
	blockingQueue.add(i);
}

抛异常：

在这里插入图片描述

4、Demo

创建一个阻塞队列，开两个线程分别对这个队列进行写个读，采用上面阻塞列的那一组方法，演示下生产消费者模式：

public class BlockQueueDemo {

    public static void main(String[] args) throws InterruptedException {
        //创建阻塞队列
        BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<>(3);

		//生产者线程
        new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                try {
                    blockingQueue.put(i);
                    System.out.println(Thread.currentThread().getName() + "线程写数据成功：" + i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        }).start();

		//消费者线程
        new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                try {
                    //每次取之前歇3秒，模拟生产快，消费慢的场景
                    TimeUnit.SECONDS.sleep(3);
                    Integer takeValue = blockingQueue.take();
                    System.out.println(Thread.currentThread().getName() + "线程取数据成功：" + takeValue);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}