Java 阻塞队列

女爷i 2023-10-09 14:13 20阅读 0赞

#### 文章目录 ####

*  BlockingQueue
 *  BlockingQueue接口
 *   *  ArrayBlockingQueue：有界队列
     *  LinkedBlockingQueue：无界队列
     *  SynchronousQueue:同步队列
 *  队列对比

## BlockingQueue ##

> 对于Queue而言，BlockingQueue是主要的线程安全的版本，具有阻塞功能，可以允许添加、删除元素被阻塞，直到成功为止，BlockingQueue相对于Queue而言增加了两个方法put、take元素。  
> ![在这里插入图片描述][ab09e65a328e4334ac0391244bef0686.png]

## BlockingQueue接口 ##

属于并发容器中的接口，在java.util.concurrent包路径下  
**BlockingQueue**不接受null元素，加入尝试通过add、put、offer等添加一个null元素时，某些实现上会抛出nullpointExeception问题。  
BlockingQueue是可以指定容量，如果给定的数据超过给定容量，便无法添加元素，如果没有指定容量约束，最大大小是**Interger.MAX\_VALUE**值  
BlockingQueue实现类主要用于**生产者-消费者**队列，另支持Collection接口。  
BlockingQueue实现了线程安全，所有排队方法都可以使用内部锁或者其他并发控制形式来达到线程安全的目的。

**三个主要实现类介绍：**

*  **ArrayBlockingQueue**：**有界阻塞队列**
 *  **LinkedBlockingQueue**：**无界阻塞队列**
 *  **SynchronousQueue**: **同步队列**

### ArrayBlockingQueue：有界队列 ###

`ArrayBlockingQueue`有界队列底层实现是数组，数组大小是固定的，假如数组一端为头，另一端为尾，那么头和尾构建一个**FIFO**队列

**属性和默认值：**

//存储的数据 存放在数组中
        final Object[] items;
        //读数据位置
        int takeIndex;
        //写入数据位置
        int putIndex;
        //数据数量
        int count;
        //队列同步相关属性
        final ReentrantLock lock;
        private final Condition notEmpty;
        private final Condition notFull;

> 通过`ArrayBlockingQueue`数据结构可知：首先是有一个数组`T[],`用来存储所有的元素，由于`ArrayBlockingQueue`最终设置为一个不可扩展大小的`Queue`，所以这里items就是初始化就固定大小的数组（final）,另外有两个索引，**头索引**`takeIndex`，**尾索引**`putIndex`，一个队列的大小`count`，要阻塞的话就必须用到一个锁和两个条件（非空，非满），这三个条件都是不可变类型。因为只有一把锁，所以任意时刻对队列只能有一个线程，意味着索引和大小的操作都是线程安全的，所以可以看到takeindex等不需要原子操作和volatile语义了。

**构造函数：**

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();
        }
    
        //通过初始容量capacity、公平性标志fair和集合c
        public ArrayBlockingQueue(int capacity, boolean fair,Collection<? extends E> c) {
        
            this(capacity, fair);
    
            final ReentrantLock lock = this.lock;
            lock.lock(); // Lock only for visibility, not mutual exclusion
            try {
        
                int i = 0;
                try {
        
                    for (E e : c) {
        
                        //数据是不能为null
                        checkNotNull(e);
                        items[i++] = e;
                    }
                } catch (ArrayIndexOutOfBoundsException ex) {
        
                    throw new IllegalArgumentException();
                }
                count = i;
                putIndex = (i == capacity) ? 0 : i;
            } finally {
        
                lock.unlock();
            }
        }

**put操作**

**可阻塞的添加元素**

public void put(E e) throws InterruptedException {
        
            //检测插入数据不能为null
            checkNotNull(e);
            //添加可中断的锁
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
        
                while (count == items.length) //容量满了需要阻塞
                    notFull.await();
                //当前集合未满，执行插入操作
                insert(e);
            } finally {
        
                //释放锁
                lock.unlock();
            }
        }
    
        private void insert(E x) {
        
            items[putIndex] = x;
            putIndex = inc(putIndex);
            ++count;
            //通知take操作已经有数据吗，如果有take方法阻塞，此时可被唤醒来执行take操作
            notEmpty.signal();
        }
    
      //循环数组的特殊标志处理 ，如果是到最大值则重定向到0号索引
        final int inc(int i) {
        
            return (++i == items.length) ? 0 : i;
        }

**插入操作，在队列满的情况下会阻塞，直到有数据take出队列时才能结束阻塞，将当前数据插入队列。**

**take方法**

**将数据从队列中移除**

public E take() throws InterruptedException {
        
            //添加可中断的锁
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
        
                while (count == 0) //队列中没有数据时，需要阻塞，直到有数据put进入队列通知该操作可以继续执行
                    notEmpty.await();
                //有数据时
                return extract();
            } finally {
        
                //释放锁
                lock.unlock();
            }
        }
    
        private E extract() {
        
            final Object[] items = this.items;
            E x = this.<E>cast(items[takeIndex]);
            items[takeIndex] = null;
            takeIndex = inc(takeIndex);
            --count;
            //发出通知 通知put方法，唤醒put操作
            notFull.signal();
            return x;
        }

> **ArrayBlockingQueue特点：**  
> 1、底层数据结构是数组，且数组大小一旦确定不可更改  
> 2、不能存储null  
> 3、阻塞功能是通过一个锁和两个隶属于该锁的Condition进行通信完成阻塞

### LinkedBlockingQueue：无界队列 ###

`LinkedBlockingQueue`有两个`lock`锁和两个`Condition`以及用于计数的`AtomicInteger`  
底层数据结构是链表，都是采用头尾节点，每个节点执行下一个节点的结构  
数据存储在Node结构中。  
引入两把锁，一个入队列锁，一个出队列的锁。满足同时有一个队列不满的Condition和一个队列不空的Condition。  
**为什么使用两把锁，一把锁是否可以？**  
一把锁完全可以的，一把锁意味着入队列和出队列同时只能有一个在进行，另一个必须等待释放锁，而从实际实现上来看，head和last是分离的，相互独立的，入队列实现是不会修改出队列的数据的，同理，出队列时也不会修改入队列的数据，这两个操作实际是相互独立，这个锁相当于两个写入锁，入队列是一种写操作，操作head，出队列是一种写操作，操作的是tail，这两是无关的。

### SynchronousQueue:同步队列 ###

`SynchronousQueue`为同步队列：每个插入操作必须等待另一个线程的移除操作，同样，任何一个移除操作都要等待另一个线程的插入操作，因此此队列中其实没有任何一个数据，或者说容量为0，SynchronousQueue更像一个管道，不像容器，资源从一个方向快速的传递到另一个方向。

## 队列对比 ##

![在这里插入图片描述][ecc66deff74d428e9921556f9a03523a.png]  
**如果不需要阻塞队列，优先选择ConcurrentLinkedQueue；  
如果需要阻塞队列，队列大小固定优先选择ArrayBlockingQueue；  
队列大小不固定优先选择LinkedBlockingQueue；  
如果需要对队列进行排序，选择PriorityBlockingQueue；  
如果需要一个快速交换的队列，选择SynchronousQueue；  
如果需要对队列中的元素进行延时操作，则选择DelayQueue。**

[ab09e65a328e4334ac0391244bef0686.png]: https://img-blog.csdnimg.cn/ab09e65a328e4334ac0391244bef0686.png
[ecc66deff74d428e9921556f9a03523a.png]: https://img-blog.csdnimg.cn/ecc66deff74d428e9921556f9a03523a.png