阻塞队列BlockingQueue-Toy模板网

这篇具有很好参考价值的文章主要介绍了阻塞队列BlockingQueue。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

阻塞队列BlockingQueue

队列

先进先出的数据结构，允许出队的一端称为队头，允许入队的一端称为队尾。

阻塞队列BlockingQueue

在java中为队列定义了一个接口规范Queue接口

public interface Queue<E> extends Collection<E> {
   
    //添加一个元素，添加成功返回true, 如果队列满了，就会抛出异常
    boolean add(E e);
    
    //添加一个元素，添加成功返回true, 如果队列满了，返回false
    boolean offer(E e);
    
    //返回并删除队首元素，队列为空则抛出异常
    E remove();
    
    //返回并删除队首元素，队列为空则返回null
    E poll();
    
    //返回队首元素，但不移除，队列为空则抛出异常
    E element();
    
    //获取队首元素，但不移除，队列为空则返回null
    E peek();
}

阻塞队列

阻塞队列提供了线程安全的队列访问方式。多线程入队互斥，多线程出队互斥，队列满了阻塞生产者线程，队列空了阻塞消费者线程

阻塞队列的接口BlockingQueue 它继承了 Queue接口。并在原有Queue接口定义的方法规范下，再新增了两个关于阻塞的方法规范。

public interface BlockingQueue<E> extends Queue<E> {
   

    // 阻塞方式 入队
    void put(E e) throws InterruptedException;

    // 阻塞方式 出队
    E take() throws InterruptedException;

   ......

}

方法	抛出异常	返回特定值	阻塞	阻塞特定时间
入队	add(e)	offer(e)	put(e)	offer(e, time, unit)
出队	remove()	poll()	take()	poll(time, unit)
获取队首元素	element()	peek()	不支持	不支持

阻塞队列的应用场景

线程池
生产者-消费者模型
消息队列
缓存系统
并发任务处理

BlockingQueue 接口的实现类都被放在了 juc 包中，它们的区别主要体现在存储结构上或对元素操作上的不同，但是对于take与put操作的原理却是类似的。

队列	描述
ArrayBlockingQueue	基于数组结构实现的一个有界阻塞队列
LinkedBlockingQueue	基于链表结构实现的一个无界阻塞队列，指定容量为有界阻塞队列
PriorityBlockingQueue	支持按优先级排序的无界阻塞队列
DelayQueue	基于优先级队列（PriorityBlockingQueue）实现的无界阻塞队列
SynchronousQueue	不存储元素的阻塞队列
LinkedTransferQueue	基于链表结构实现的一个无界阻塞队列
LinkedBlockingDeque	基于链表结构实现的一个双端阻塞队列

详情

ArrayBlockingQueue

简介

ArrayBlockingQueue它队列使用的数据结构是双指针的环形数组，入队出队是基于生产者消费者模型实现的，它入队和出队使用的是同一把锁。

基本使用

BlockingQueue queue = new ArrayBlockingQueue(1024);
queue.put("1");   //向队列中添加元素
Object object = queue.take();   //从队列中取出元素

其实ava.util.concurrent.locks.Condition接口的注释中给出的案例就和ArrayBlockingQueue的实现很相似，如下所示使用一个环形数组实现生产者消费者模式：文章来源地址https://www.toymoban.com/news/detail-406720.html

class BoundedBuffer {
   
    final Lock lock = new ReentrantLock();
    final Condition notFull  = lock.newCondition(); 
    final Condition notEmpty = lock.newCondition(); 

    final Object[] items = new Object[100];
    int putptr, takeptr, count;

    // 生产者，把方法的入参往队列中存，队列是一个环形数组
    public void put(Object x) throws InterruptedException {
   
        lock.lock();
        try {
   
            
            // 队列放满了生产者就阻塞
            while (count == items.length)
                notFull.await();
            
            // 往队列存数据
            items[putptr] = x;
            if (++putptr == items.length) putptr = 0;
            
            // 队列中新存值了，要通知消费者开始消费了
            ++count;
            notEmpty.signal();
        } finally {
   
            lock.unlock();
        }
    }

    // 消费者，从队列中取
    public Object take() throws InterruptedException {
   
        lock.lock();
        try {
   
            
            // 队列空了，消费者就阻塞
            while (count == 0)
                notEmpty.await();
            
            // 从队列中取值
            Object x = items[takeptr];
            if (++takeptr == items.length) takeptr = 0;
            
            // 已经消费了，队列现在不是满的了，唤醒生产者进行生产
            --count;
            notFull.signal();
            return x;
        } finally {
   
            lock.unlock();
        }
    }