如何在Java中实现阻塞队列

如何在Java中实现阻塞队列？很多新手对此不是很清楚，为了帮助大家解决这个难题，下面小编将为大家详细讲解，有这方面需求的人可以来学习下，希望你能有所收获。

Java阻塞队列

阻塞队列和普通队列主要区别在阻塞二字：

• 阻塞添加：队列已满时，添加元素线程会阻塞，直到队列不满时才唤醒线程执行添加操作

• 阻塞删除：队列元素为空时，删除元素线程会阻塞，直到队列不为空再执行删除操作

常见的阻塞队列有 LinkedBlockingQueue 和 ArrayBlockingQueue，其中它们都实现 BlockingQueue 接口，该接口定义了阻塞队列需实现的核心方法：

public interface BlockingQueue<E> extends Queue<E> {// 添加元素到队尾，成功返回true，队列满抛出异常 IllegalStateException    boolean add(E e);// 添加元素到队尾，成功返回 true，队列满返回 false    boolean offer(E e);// 阻塞添加    void put(E e) throws InterruptedException;// 阻塞添加，包含最大等待时长    boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException;// 阻塞移除队顶元素    E take() throws InterruptedException;// 阻塞移除队顶元素，包含最大等待时长    E poll(long timeout, TimeUnit unit) throws InterruptedException;// 返回可以添加到队列不阻塞的最大数量    int remaininGCapacity();// 如果存在元素则删除，成功返回 true，失败返回 false    boolean remove(Object o);// 是否包含某元素    public boolean contains(Object o);    // 批量移除元素并添加入指定集合    int drainTo(Collection<? super E> c);// 批量移除包含最大数量    int drainTo(Collection<? super E> c, int maxElements);}

除了上面的方法，还有三个继承自 Queue 接口的方法常常被用到：

// 获取队列头元素，不删除，没有抛出异常 NoSuchElementExceptionE element();// 获取队列头元素，不删除，没有返回 nullE peek();// 获取并移除队列头元素，没有返回 nulE poll();

根据具体作用，方法可以被分为以下三类：

• 添加元素类：add() 成功返回 true，失败抛异常、offer() 成功返回 true，失败返回 false，可以定义最大等待时长、put() 阻塞方法

• 删除元素类：remove() 成功返回 true，失败返回 false、poll() 成功返回被移除元素，为空返回 null、take() 阻塞方法

• 查询元素类：element() 成功返回元素，否则抛出异常、peek() 返回对应元素或 null

根据方法类型又可以分为阻塞和非阻塞，其中 put()、take() 是阻塞方法，带最大等待时长的 offer() 和 poll() 也是阻塞方法，其余都是非阻塞方法，阻塞队列基于上述方法实现

ArrayBlockingQueue 基于数组实现，满足队列先进先出特性，下面我们通过一段代码初步认识：

public class ArrayBlockingQueueTest {    ArrayBlockingQueue<TestProduct> queue = new ArrayBlockingQueue<TestProduct>(1);    public static void main(String[] args) {        ArrayBlockingQueueTest test = new ArrayBlockingQueueTest();        new Thread(test.new Product()).start();        new Thread(test.new Customer()).start();    }    class Product implements Runnable {        @Override        public void run() {            while (true) {                try {                    queue.put(new TestProduct());                    System.out.println("生产者创建产品等待消费者消费");                } catch (InterruptedException e) {                    e.printStackTrace();                }            }        }    }    class Customer implements Runnable {        @Override        public void run() {            while (true) {                try {                    Thread.sleep(1000);                    queue.take();                    System.out.println("消费者消费产品等待生产者创建");                } catch (InterruptedException e) {                    e.printStackTrace();                }            }        }    }    class TestProduct {    }}

上述代码比较简单，在一个容量为1的阻塞队列中，生产者和消费者由于容量限制依次阻塞运行。

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

从代码可以看出，ArrayBlockingQueue 通过一个 ReentrantLock 锁以及两个 Condition 等待队列实现，它的属性如下：

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;// 不为空，用于take()操作    private final Condition notEmpty;// 不满，用于put()操作    private final Condition notFull;// 迭代器    transient Itrs itrs = null;}

从代码可以看出，ArrayBlockingQueue 使用同一个锁、移除元素和添加元素通过数组下标的方式记录，分表表示队列头和队列尾。通过两个等待队列分别阻塞 take() 和 put() 方法，下面我们直接看源码：

public boolean add(E e) {    if (offer(e))        return true;    else        throw new IllegalStateException("Queue full");}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++;    // 唤醒take线程    notEmpty.signal();}

从代码可以看出：add() 方法基于 offer() 方法实现，offer() 方法添加失败返回 false 后，add() 方法抛出异常。offer() 方法会加锁，保证线程安全，队列没满时执行入队操作，入队操作通过操作数组实现，并且通过循环复用数组空间。元素添加成功后队列不为空，调用 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();}}

从代码可以看出，当队列满时，当前线程会被挂起到等待队列中，直到队列不满时被唤醒执行添加操作。下面我们看删除操作：

public boolean remove(Object o) {// 判断是否为 NULL    if (o == null) return false;    final Object[] items = this.items;    final ReentrantLock lock = this.lock;    lock.lock();    try {        if (count > 0) {            final int putIndex = this.putIndex;            int i = takeIndex;            // 从移除下标开始遍历到添加新元素的下标            do {                if (o.equals(items[i])) {                    removeAt(i);                    return true;                }                // 循环判断，移除下标可能大于添加下标（添加下标二次遍历时）                if (++i == items.length)                    i = 0;             } while (i != putIndex);        }        return false;    } finally {        lock.unlock();    }}void removeAt(final int removeIndex) {final Object[] items = this.items;// 要删除的元素正好是移除下标if (removeIndex == takeIndex) {    items[takeIndex] = null;    // 循环删除    if (++takeIndex == items.length)        takeIndex = 0;    count--;    if (itrs != null)        itrs.elementDequeued();} else {    final int putIndex = this.putIndex;    // 如果不是移除下标，从该下标开始到添加下标，所有元素左移一位    for (int i = removeIndex;;) {        int next = i + 1;        if (next == items.length)            next = 0;        if (next != putIndex) {        // 向左移除            items[i] = items[next];            i = next;        } else {        // 最后put下标置为null            items[i] = null;            this.putIndex = i;            break;        }    }    count--;    // 更新迭代器    if (itrs != null)        itrs.removedAt(removeIndex);}notFull.signal();}

remove() 和 poll()、take() 不同，它可以删除指定的元素。这里需要考虑删除的元素不是移除索引指向的情况，从代码可以看出，当要删除的元素不是移除索引指向的元素时，将所有从被删除元素下标开始到添加元素下标所有元素左移一位。

public E poll() {final ReentrantLock lock = this.lock;lock.lock();try {    return (count == 0) ? null : dequeue();} finally {    lock.unlock();}}private E dequeue() {final Object[] items = this.items;E x = (E) items[takeIndex];items[takeIndex] = null;if (++takeIndex == items.length)    takeIndex = 0;count--;if (itrs != null)    itrs.elementDequeued();// 移除元素后唤醒put()添加线程notFull.signal();return x;}

相比 remove() 方法，poll() 方法简单了很多，这里不做赘述，下面我们看 take()：

public E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {// 队列为空就挂起    while (count == 0)        notEmpty.await();    return dequeue();} finally {    lock.unlock();}}

take() 方法和 put() 方法可以说基本一致，相对也比较简单，最后我们来看看两个查询方法：

public E element() {    E x = peek();    if (x != null)        return x;    else        throw new NoSuchElementException();}public E peek() {    final ReentrantLock lock = this.lock;    lock.lock();    try {    // 直接返回移除元素下标对应的元素，也就是队列头        return itemAt(takeIndex);     } finally {        lock.unlock();    }}final E itemAt(int i) {    return (E) items[i];}

element() 基于 peek() 方法实现实现、当队列为空时，peek() 方法返回 null，element() 抛出异常。关于 ArrayBlockingQueue 就介绍到这里

LinkedBlockingQueue 基于链表实现，它的属性如下：

public class LinkedBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {// 链表节点，存储元素    static class node<E> {        E item;        Node<E> next;        Node(E x) { item = x; }    }// 链表容量    private final int capacity;// 当前元素数量    private final AtomicInteger count = new AtomicInteger();// 头节点    transient Node<E> head;    // 尾节点    private transient Node<E> last;// 删除锁    private final ReentrantLock takeLock = new ReentrantLock();// 不为空等待队列    private final Condition notEmpty = takeLock.newCondition();// 添加锁    private final ReentrantLock putLock = new ReentrantLock();// 不满等待队列    private final Condition notFull = putLock.newCondition();}

从代码可以看出，元素被封装为 Node 节点保存在单向链表中，其中链表默认长度为 Integer.MAX_VALUE，因此在使用时需注意内存溢出：当添加元素速度大于删除元素速度时，队列最终会记录到大量不会用到并且无法回收的对象，导致内存溢出。

ArrayBlockingQueue 和 LinkedBlockingQueue 的主要区别在于 ReentrantLock 锁的数量和等待队列，LinkedBlockingQueue 用到两个锁和两个等待队列，也就是说添加和删除操作可以并发执行，整体效率更高。下面我们直接看代码：

public boolean add(E e) {     if (offer(e))         return true;     else         throw new IllegalStateException("Queue full");}public boolean offer(E e) {// 元素为空抛出异常if (e == null) throw new NullPointerException();// 获取当前队列容量final AtomicInteger count = this.count;// 队列已满时直接返回falseif (count.get() == capacity)    return false;int c = -1;Node<E> node = new Node<E>(e);// 获取添加锁final ReentrantLock putLock = this.putLock;putLock.lock();try {// 二次判断，因为上面判断时未加锁，数据可能已更新    if (count.get() < capacity) {    // 入队操作        enqueue(node);        // 获取还未添加元素前，队列的容量        c = count.getAndIncrement();        if (c + 1 < capacity)        // 唤醒其它添加元素的线程            notFull.signal();    }} finally {    putLock.unlock();}// 如果添加前队列没有数据，也就是说现在有一条数据时if (c == 0)// 唤醒take线程   signalNotEmpty();return c >= 0;}private void enqueue(Node<E> node) {     last = last.next = node;}private void signalNotEmpty() {// 唤醒take线程前必须获取对应take锁    final ReentrantLock takeLock = this.takeLock;    takeLock.lock();        notEmpty.signal();    } finally {        takeLock.unlock();    }}

这里有以下几点需要我们注意：

LinkedBlockingQueue count 属性必须通过并发类封装，因为可能存在添加、删除两个线程并发执行，需考虑同步

这里需要判断两次的主要原因在于方法开始时并没有加锁，数值可能改变，因此在获取到锁后需要二次判断

和 ArrayBlockingQueue 不同，LinkedBlockingQueue 在队列不满时会唤醒添加线程，这样做的原因是 LinkedBlockingQueue 中添加和删除操作使用不同的锁，各自只需管好自己，还可以提高吞吐量。而 ArrayBlockingQueue 使用唯一锁，这样做会导致移除线程永远不被唤醒或添加线程永远不被唤醒，吞吐量较低

添加元素前队列长度为0才唤醒移除线程，因为队列长度为0时，移除线程肯定已经挂起，此时唤醒一个移除线程即可。因为移除线程和添加线程类似，都会自己唤醒自己。而 c>0 时只会有两种情况：存在移除线程在运行，如果有会递归唤醒，无须我们参与、不存在移除线程运行，此时也无须我们参与，等待调用 take()、poll() 方法即可

唤醒只针对 put()、take() 方法阻塞的线程，offer() 方法直接返回（不包含最大等待时长），不参与唤醒场景

下面我们来看 put() 阻塞方法的实现：

public void put(E e) throws InterruptedException {    if (e == null) throw new NullPointerException();    int c = -1;    Node<E> node = new Node<E>(e);    final ReentrantLock putLock = this.putLock;    final AtomicInteger count = this.count;    putLock.lockInterruptibly();    try {    // 队列满时阻塞        while (count.get() == capacity) {            notFull.await();        }        // 入队        enqueue(node);        c = count.getAndIncrement();        if (c + 1 < capacity)            notFull.signal();    } finally {        putLock.unlock();    }    if (c == 0)        signalNotEmpty();}

从代码可以看出，put() 方法和 offer() 方法唯一区别在于自身通过 condition 阻塞挂起到等待队列，其余基本相同。至此关于添加操作介绍完毕，下面我们看移除方法：

public boolean remove(Object o) {if (o == null) return false;// 同时加两个锁fullyLock();try {// 循环查找    for (Node<E> trail = head, p = trail.next; p != null; trail = p, p = p.next) {        if (o.equals(p.item)) {            unlink(p, trail);            return true;        }    }    return false;} finally {    fullyUnlock();}}void unlink(Node<E> p, Node<E> trail) {// p是要溢出的节点，trail是它的前驱节点// 方便gc    p.item = null;    // 引用取消    trail.next = p.next;    if (last == p)        last = trail;    if (count.getAndDecrement() == capacity)        notFull.signal();}void fullyLock() {    putLock.lock();    takeLock.lock();}void fullyUnlock() {    takeLock.unlock();    putLock.unlock();}

从代码可以看出，remove() 方法只会在操作前容量不满时唤醒创建线程，并不会唤醒移除线程。并且由于我们不确定要删除元素的位置，因此此时需要加两个锁，确保数据安全。

public E poll() {    final AtomicInteger count = this.count;    if (count.get() == 0)        return null;    E x = null;    int c = -1;    final ReentrantLock takeLock = this.takeLock;    takeLock.lock();    try {        if (count.get() > 0) {            x = dequeue();            // 获取移除前队列的元素数量            c = count.getAndDecrement();            if (c > 1)                notEmpty.signal();        }    } finally {        takeLock.unlock();    }    // 移除前如果队列是满的，唤醒添加线程    if (c == capacity)        signalNotFull();    return x;}private E dequeue() {Node<E> h = head;// 获取要删除的节点Node<E> first = h.next; // 清除原来的头结点(方便gc)h.next = h; // 设置新的头结点head = first;// 获取返回值E x = first.item;// 新头结点置为空first.item = null;return x;}

需要注意的一点，每次出队时更换 head 节点，head 节点本身不保存数据，head.next 记录下次需要出队的元素，每次出队后 head.next 变为新的 head 节点返回并置为 null

poll() 方法和上面提到的 offer() 方法基本镜像相同，这里我再不做过多赘述

public E take() throws InterruptedException {    E x;    int c = -1;    final AtomicInteger count = this.count;    final ReentrantLock takeLock = this.takeLock;    takeLock.lockInterruptibly();    try {    // 队列为空就挂起        while (count.get() == 0) {            notEmpty.await();        }        x = dequeue();        c = count.getAndDecrement();        if (c > 1)            notEmpty.signal();    } finally {        takeLock.unlock();    }    if (c == capacity)        signalNotFull();    return x;}

take() 方法和 poll() 方法类似，区别在于新增了阻塞逻辑。至此关于溢出元素方法介绍完毕，最后我们看看查询方法源码：

public LinkedBlockingQueue(int capacity) {   if (capacity <= 0) throw new IllegalArgumentException();   this.capacity = capacity;   last = head = new Node<E>(null);}public E element() {    E x = peek();    if (x != null)        return x;    else        throw new NoSuchElementException();}public E peek() {    if (count.get() == 0)        return null;    final ReentrantLock takeLock = this.takeLock;    takeLock.lock();    try {        Node<E> first = head.next;        if (first == null)            return null;        else            return first.item;    } finally {        takeLock.unlock();    }}

从代码可以看出，默认 head 和 last 头尾节点都为 null，入队时直接从 next 开始操作，也就是说 head 节点不保存数据。

最后我们来看看有最大等待时长的 offer() 方法：

public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {if (e == null) throw new NullPointerException();// 将时间转换成纳秒long nanos = unit.toNanos(timeout);int c = -1;// 获取锁final ReentrantLock putLock = this.putLock;// 获取当前队列大小final AtomicInteger count = this.count;// 可中断锁putLock.lockInterruptibly();try {    while (count.get() == capacity) {    // 小于0说明已到达最大等待时长        if (nanos <= 0)            return false;        // 如果队列已满，根据等待队列阻塞等待        nanos = notFull.awaitNanos(nanos);    }    // 队列没满直接入队    enqueue(new Node<E>(e));    c = count.getAndIncrement();    if (c + 1 < capacity)        notFull.signal();} finally {     putLock.unlock();}if (c == 0)    signalNotEmpty();return true;} public final long awaitNanos(long nanosTimeout) throws InterruptedException {    if (Thread.interrupted()) throw new InterruptedException();// 将当前线程封装为 AQS Node 类加入等待队列    Node node = addConditionWaiter();    // 释放锁    int savedState = fullyRelease(node);    //计算过期时间    final long deadline = System.nanoTime() + nanosTimeout;    int interruptMode = 0;    // 当前线程没有唤醒进入同步队列时    while (!isOnSyncQueue(node)) {    // 已经等待相应时间，删除当前节点，将状态设置为已关闭从队列删除        if (nanosTimeout <= 0L) {            transferAfterCancelledWait(node);            break;        }        // 判断是否超时        if (nanosTimeout >= spinForTimeoutThreshold)        // 挂起线程            LockSupport.parkNanos(this, nanosTimeout);        // 判断线程状态是否被中断        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)            break;        // 重新计算剩余等待时间        nanosTimeout = deadline - System.nanoTime();    }    // 被唤醒后执行自旋操作争取获得锁，同时判断线程是否被中断    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)        interruptMode = REINTERRUPT;    if (node.nextWaiter != null)    // 清理等待队列中不为Condition状态的线程        unlinkCancelledWaiters();    // 判断是否被中断    if (interruptMode != 0)    // 抛出异常或中断线程，独占模式抛出异常，共享模式中断线程        reportInterruptAfterWait(interruptMode);    // 返回时差,如果成功当前时间小于最大等待时长，返回值大于0，否则返回值小于0    return deadline - System.nanoTime();}

从代码可以看出，包含最大等待时长的 offer()、poll() 方法通过循环判断时间是否超时的方式挂起在等待队列，达到最大等待时长还未被唤醒或没被执行就返回

ArrayBlockingQueue 和 LinkedBlockingQueue 对比：

• 大小不同，一个有界，一个无界。ArrayBlockingQueue 必须指定初始大小，LinkedBlockingQueue 无界时可能内存溢出

• 一个采用数组，一个采用链表，数组保存无须创建新对象，链表需创建 Node 对象

• 锁机制不同，ArrayBlockingQueue 添加删除操作使用同一个锁，两者操作不能并发执行。LinkedBlockingQueue 添加和删除使用不同锁，添加和删除操作可并发执行，整体效率 LinkedBlockingQueue 更高

看完上述内容是否对您有帮助呢？如果还想对相关知识有进一步的了解或阅读更多相关文章，请关注编程网精选频道，感谢您对编程网的支持。