阻塞队列，有界队列，无界队列

骑猪看日落 2022-11-07 13:29 195阅读 0赞

在并发编程中，有时候需要使用线程安全的队列。如果要实现一个线程安全的队列有两种方式：一种是使用阻塞算法，另一种是使用非阻塞算法。

使用阻塞算法的队列可以用一个锁（入队和出队用同一把锁）或两个锁（入队和出队用不同的锁）等方式来实现。非阻塞的实现方式则可以使用自旋+CAS的方式来实现。

# 基本概念 #

## 阻塞队列与非阻塞队列 ##

阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。这两个附加的操作支持阻塞的插入和移除方法。

*  支持阻塞的插入方法offer：当队列满时，队列会阻塞插入元素的线程，直到队列不满。
 *  支持阻塞的移除方法take：在队列为空时，获取元素的线程会等待队列变为非空。

阻塞队列常用于生产者和消费者的场景，生产者是向队列里添加元素的线程，消费者是从队列里取元素的线程。阻塞队列就是生产者用来存放元素、消费者用来获取元素的容器。

非阻塞队列：若队列为空从中获取元素则会返回空，若队列满了插入元素则会抛出异常。

## 有界队列与无界队列 ##

*  有界队列：就是有固定大小的队列。比如设定了固定大小的ArrayBlockingQueue，又或者大小为0，只是在生产者和消费者中做中转用的SynchronousQueue。
 *  无界队列：指的是没有设置固定大小的队列。这些队列的特点是可以直接入列，直到溢出。当然现实几乎不会有到这么大的容量（超过 Integer.MAX\_VALUE），所以从使用者的体验上，就相当于 “无界”。

# Java中的阻塞队列 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]

*  ArrayBlockingQueue：一个由数组结构组成的有界阻塞队列。
 *  LinkedBlockingQueue：一个由链表结构组成的有界阻塞队列。
 *  PriorityBlockingQueue：一个支持优先级排序的无界阻塞队列。
 *  DelayQueue：一个使用优先级队列实现的延迟无界阻塞队列。
 *  SynchronousQueue：一个不存储元素的阻塞队列。
 *  LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。
 *  LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。

另外还有一个DelayedWorkQueue，是ScheduledThreadPoolExecutor的内部类。

## 阻塞队列中的方法 ##

<table> 
 <thead> 
  <tr> 
   <th>方法类型</th> 
   <th>抛出异常</th> 
   <th>特殊值</th> 
   <th>阻塞</th> 
   <th>超时</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>插入时队列满了的表现</td> 
   <td>add（IllegalStateException）</td> 
   <td>offer（false）</td> 
   <td>put</td> 
   <td>offer</td> 
  </tr> 
  <tr> 
   <td>移除时队列为空的表现</td> 
   <td>remove（NoSuchElementException）</td> 
   <td>poll（null）</td> 
   <td>take</td> 
   <td>poll</td> 
  </tr> 
  <tr> 
   <td>检查时队列为空的表现</td> 
   <td>element（NoSuchElementException）</td> 
   <td>peek（null）</td> 
   <td>不可用</td> 
   <td>不可用</td> 
  </tr> 
 </tbody> 
</table>

## ArrayBlockingQueue ##

基于数组实现的有界阻塞队列，ArrayBlockingQueue内部维护这一个定长数组，阻塞队列的大小在初始化时就已经确定了，其后无法更改。

ArrayBlockingQueue支持公平性和非公平性，默认采用非公平模式，可以通过构造函数设置为公平访问策略。

## PriorityBlockingQueue ##

PriorityBlockingQueue是支持优先级的无界队列。默认情况下采用自然顺序排序，当然也可以通过自定义Comparator来指定元素的排序顺序，不能保证同优先级元素的顺序。

PriorityBlockingQueue内部采用二叉堆的实现方式。

## DelayQueue ##

DelayQueue是一个支持延时操作的无界阻塞队列。列头的元素是最先“到期”的元素，如果队列里面没有元素到期，是不能从列头获取元素的，哪怕有元素也不行。也就是说只有在延迟期满时才能够从队列中去元素。

它主要运用于如下场景：

*  缓存系统的设计：缓存是有一定的时效性的，可以用DelayQueue保存缓存的有效期，然后利用一个线程查询DelayQueue，如果取到元素就证明该缓存已经失效了。
 *  定时任务的调度：DelayQueue保存当天将要执行的任务和执行时间，一旦取到元素（任务），就执行该任务。

DelayQueue采用支持优先级的PriorityQueue来实现，但是队列中的元素必须要实现Delayed接口，Delayed接口用来标记那些应该在给定延迟时间之后执行的对象，该接口提供了getDelay()方法返回元素节点的剩余时间。同时，元素也必须要实现compareTo()方法，compareTo()方法需要提供与getDelay()方法一致的排序。

## SynchronousQueue ##

SynchronousQueue是一个神奇的队列，他是一个不存储元素的阻塞队列，也就是说他的每一个put操作都需要等待一个take操作，否则就不能继续添加元素了，有点儿像Exchanger，类似于生产者和消费者进行交换。

队列本身不存储任何元素，所以非常适用于传递性场景，两者直接进行对接。其吞吐量会高于ArrayBlockingQueue和LinkedBlockingQueue。

SynchronousQueue支持公平和非公平的访问策略，在默认情况下采用非公平性，也可以通过构造函数来设置为公平性。

SynchronousQueue的实现核心为Transferer接口，该接口有TransferQueue和TransferStack两个实现类，分别对应着公平策略和非公平策略。接口Transferer有一个tranfer()方法，该方法定义了转移数据，如果e != null，相当于将一个数据交给消费者，如果e == null，则相当于从一个生产者接收一个消费者交出的数据。

## LinkedTransferQueue ##

LinkedTransferQueue与其他BlockingQueue相比，他多实现了一个接口TransferQueue，该接口是对BlockingQueue的一种补充，多了tryTranfer()和transfer()两类方法：

*  tranfer()：若当前存在一个正在等待获取的消费者线程，即立刻移交之。 否则，会插入当前元素e到队列尾部，并且等待进入阻塞状态，到有消费者线程取走该元素。
 *  tryTranfer()： 若当前存在一个正在等待获取的消费者线程（使用take()或者poll()函数），使用该方法会即刻转移/传输对象元素e；若不存在，则返回false，并且不进入队列。这是一个不阻塞的操作。

## LinkedBlockingDeque ##

LinkedBlockingDeque是一个有链表组成的双向阻塞队列，与前面的阻塞队列相比它支持从两端插入和移出元素。以first结尾的表示从对头操作，以last结尾的表示从对尾操作。

在初始化LinkedBlockingDeque时可以初始化队列的容量，用来防止其再扩容时过渡膨胀。另外双向阻塞队列可以运用在“工作窃取”模式中。

# ArrayBlockingQueue的源码分析 #

ArrayBlockingQueue，一个由数组实现的有界阻塞队列。该队列采用FIFO的原则对元素进行排序添加的。

## 数据结构 ##

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
            implements BlockingQueue<E>, java.io.Serializable { 
    
        final Object[] items; // 存储元素的数组
    
        int takeIndex; // 队首位置
    
        int putIndex; // 队尾位置
    
        int count; // 元素的个数
    
        final ReentrantLock lock; // 锁，出列入列都必须获取该锁，两个步骤共用一个锁
    
        private final Condition notEmpty; // 出队条件
    
        private final Condition notFull; // 入队条件

ArrayBlockingQueue内部使用可重入锁ReentrantLock + Condition来完成多线程环境的并发操作。

## 构造方法 ##

// 指定初始化时元素个数，可见是个有界队列
    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 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) { 
                    checkNotNull(e);
                    items[i++] = e;
                }
            } catch (ArrayIndexOutOfBoundsException ex) { 
                throw new IllegalArgumentException();
            }
            count = i;
            putIndex = (i == capacity) ? 0 : i;
        } finally { 
            lock.unlock();
        }
    }

## offer() ##

public boolean offer(E e) { 
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try { 
            if (count == items.length)
                // 队满了
                return false;
            else { 
                enqueue(e);
                return true;
            }
        } finally { 
            lock.unlock();
        }
    }
    
    // 插入队尾
    private void enqueue(E x) { 
        final Object[] items = this.items;
        items[putIndex] = x;
        if (++putIndex == items.length)
            putIndex = 0;
        count++;
        notEmpty.signal();
    }

## put() ##

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();
        }
    }

## take() ##

public E take() throws InterruptedException { 
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try { 
            while (count == 0)
                notEmpty.await(); // 队空等待
            return dequeue(); // 出队
        } finally { 
            lock.unlock();
        }
    }
    
    private E dequeue() { 
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        E x = (E) items[takeIndex];
        items[takeIndex] = null;
        if (++takeIndex == items.length)
            takeIndex = 0;
        count--;
        if (itrs != null)
            itrs.elementDequeued();
        notFull.signal();
        return x;
    }

## poll() ##

public E poll() { 
        final ReentrantLock lock = this.lock;
        lock.lock();
        try { 
            // 队空返回null，不等待
            return (count == 0) ? null : dequeue();
        } finally { 
            lock.unlock();
        }
    }

# 手写阻塞队列 #

package com.morris.concurrent.pattern.guardedsuspension;
    
    import java.util.ArrayList;
    
    public class ArrayListBlockingQueue<T> { 
    
        private ArrayList<T> taskList;
    
        private int queueSize;
    
        public ArrayListBlockingQueue(int queueSize) { 
            this.queueSize = queueSize;
            this.taskList = new ArrayList<>(queueSize);
        }
    
        public void offer(T t) throws InterruptedException { 
            synchronized (this.taskList) { 
                while(this.taskList.size() == queueSize) { 
                    this.taskList.wait();
                }
                this.taskList.add(t);
                this.taskList.notifyAll();
            }
        }
    
        public T take() throws InterruptedException { 
            synchronized (this.taskList) { 
                while (this.taskList.isEmpty()) { 
                    this.taskList.wait();
                }
                this.taskList.notifyAll();
                return this.taskList.remove(0);
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]: /images/20221023/f714d5022b1a47e9b4912246c9a98e44.png