一文带你彻底掌握阻塞队列!

2023年 12月 15日 56.1k 0

一、摘要

在之前的文章中,我们介绍了生产者和消费者模型的最基本实现思路,相信大家对它已经有一个初步的认识。

在 Java 的并发包里面还有一个非常重要的接口:BlockingQueue。

BlockingQueue是一个阻塞队列,更为准确的解释是:BlockingQueue是一个基于阻塞机制实现的线程安全的队列。通过它也可以实现生产者和消费者模型,并且效率更高、安全可靠,相比之前介绍的生产者和消费者模型,它可以同时实现生产者和消费者并行运行。

那什么是阻塞队列呢?

简单的说,就是当参数在入队和出队时,通过加锁的方式来避免线程并发操作时导致的数据异常问题。

在 Java 中,能对线程并发执行进行加锁的方式主要有synchronized和ReentrantLock,其中BlockingQueue采用的是ReentrantLock方式实现。

与此对应的还有非阻塞机制的队列,主要是采用 CAS 方式来控制并发操作,例如:ConcurrentLinkedQueue,这个我们在后面的文章再进行分享介绍。

今天我们主要介绍BlockingQueue相关的知识和用法,废话不多说了,进入正题!

二、BlockingQueue 方法介绍

打开BlockingQueue的源码,你会发现它继承自Queue,正如上文提到的,它本质是一个队列接口。

public interface BlockingQueue extends Queue {
 //...省略
}

关于队列,我们在之前的集合系列文章中对此有过深入的介绍,本篇就再次简单的介绍一下。

队列其实是一个数据结构,元素遵循先进先出的原则,所有新元素的插入,也被称为入队操作,会插入到队列的尾部;元素的移除,也被称为出队操作,会从队列的头部开始移除,从而保证先进先出的原则。

在Queue接口中,总共有 6 个方法,可以分为 3 类,分别是:插入、移除、查询,内容如下:

方法描述add(e)插入元素,如果插入失败,就抛异常offer(e)插入元素,如果插入成功,就返回 true;反之 falseremove()移除元素,如果移除失败,就抛异常poll()移除元素,如果移除成功,返回 true;反之 falseelement()获取队首元素,如果获取结果为空,就抛异常peek()获取队首元素,如果获取结果为空,返回空对象

因为BlockingQueue是Queue的子接口,了解Queue接口里面的方法,有助于我们对BlockingQueue的理解。

除此之外,BlockingQueue还单独扩展了一些特有的方法,内容如下:

方法描述put(e)插入元素,如果没有插入成功,线程会一直阻塞,直到队列中有空间再继续offer(e, time, unit)插入元素,如果在指定的时间内没有插入成功,就返回 false;反之 truetake()移除元素,如果没有移除成功,线程会一直阻塞,直到队列中新的数据被加入poll(time, unit)移除元素,如果在指定的时间内没有移除成功,就返回 false;反之 truedrainTo(Collection c, int maxElements)一次性取走队列中的数据到 c 中,可以指定取的个数。该方法可以提升获取数据效率,不需要多次分批加锁或释放锁

分析源码,你会发现相比普通的Queue子类,BlockingQueue子类主要有以下几个明显的不同点:

  • 1.元素插入和移除时线程安全:主要是通过在入队和出队时进行加锁,保证了队列线程安全,加锁逻辑采用ReentrantLock实现
  • 2.支持阻塞的入队和出队方法:当队列满时,会阻塞入队的线程,直到队列不满;当队列为空时,会阻塞出队的线程,直到队列中有元素;同时支持超时机制,防止线程一直阻塞

三、BlockingQueue 用法详解

打开源码,BlockingQueue接口的实现类非常多,我们重点讲解一下其中的 5 个非常重要的实现类,分别如下表所示。

实现类功能ArrayBlockingQueue基于数组的阻塞队列,使用数组存储数据,需要指定长度,所以是一个有界队列LinkedBlockingQueue基于链表的阻塞队列,使用链表存储数据,默认是一个无界队列;也可以通过构造方法中的capacity设置最大元素数量,所以也可以作为有界队列SynchronousQueue一种没有缓冲的队列

生产者产生的数据直接会被消费者获取并且立刻消费PriorityBlockingQueue基于优先级别的阻塞队列,底层基于数组实现,是一个无界队列DelayQueue延迟队列,其中的元素只有到了其指定的延迟时间,才能够从队列中出队

下面我们对以上实现类的用法,进行一一介绍。

3.1、ArrayBlockingQueue

ArrayBlockingQueue是一个基于数组的阻塞队列,初始化的时候必须指定队列大小,源码实现比较简单,采用的是ReentrantLock和Condition实现生产者和消费者模型,部分核心源码如下:

public class ArrayBlockingQueue extends AbstractQueue
        implements BlockingQueue, java.io.Serializable {

 /** 使用数组存储队列中的元素 */
 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  1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }
}

把最上面的样例Container中的阻塞队列实现类换成LinkedBlockingQueue,调整如下:

/**
 * 初始化阻塞队列
 */
private final BlockingQueue queue = new LinkedBlockingQueue();

再次运行结果如下:

生产者: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、SynchronousQueue

SynchronousQueue是一个没有缓冲的队列,生产者产生的数据直接会被消费者获取并且立刻消费,相当于传统的一个请求对应一个应答模式。

相比ArrayBlockingQueue和LinkedBlockingQueue,SynchronousQueue实现机制也不同,它主要采用队列和栈来实现数据的传递,中间不存储任何数据,生产的数据必须得消费者处理,线程阻塞方式采用 JDK 提供的LockSupport park/unpark函数来完成,也支持公平和非公平两种模式。

  • 当采用公平模式时:使用一个 FIFO 队列来管理多余的生产者和消费者
  • 当采用非公平模式时:使用一个 LIFO 栈来管理多余的生产者和消费者,这也是SynchronousQueue默认的模式

部分核心源码如下:

public class SynchronousQueue extends AbstractQueue
    implements BlockingQueue, java.io.Serializable {

    /**不同的策略实现*/
    private transient volatile Transferer transferer;

 /**默认非公平模式*/
    public SynchronousQueue() {
        this(false);
    }

    /**可以选策略,也可以采用公平模式*/
    public SynchronousQueue(boolean fair) {
        transferer = fair ? new TransferQueue() : new TransferStack();
    }

 /**入队操作*/
    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,代码如下:

public class Container {

    /**
     * 初始化阻塞队列
     */
    private final BlockingQueue queue = new SynchronousQueue();


    /**
     * 添加数据到阻塞队列
     * @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();
        }
    }
}

再次运行结果如下:

生产者: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、PriorityBlockingQueue

PriorityBlockingQueue是一个基于优先级别的阻塞队列,底层基于数组实现,可以认为是一个无界队列。

PriorityBlockingQueue与ArrayBlockingQueue的实现逻辑,基本相似,也是采用ReentrantLock来实现加锁的操作。

最大不同点在于:

  • 1.PriorityBlockingQueue内部基于数组实现的最小二叉堆算法,可以对队列中的元素进行排序,插入队列的元素需要实现Comparator或者Comparable接口,以便对元素进行排序
  • 2.其次,队列的长度是可扩展的,不需要显式指定长度,上限为Integer.MAX_VALUE - 8

部分核心源码如下:

public class PriorityBlockingQueue extends AbstractQueue
implements BlockingQueue, java.io.Serializable {

/**队列元素*/
private transient Object[] queue;

/**比较器*/
private transient Comparator

相关文章

JavaScript2024新功能:Object.groupBy、正则表达式v标志
PHP trim 函数对多字节字符的使用和限制
新函数 json_validate() 、randomizer 类扩展…20 个PHP 8.3 新特性全面解析
使用HTMX为WordPress增效:如何在不使用复杂框架的情况下增强平台功能
为React 19做准备:WordPress 6.6用户指南
如何删除WordPress中的所有评论

发布评论