Java LinkedBlockingQueue 实现-阿里云开发者社区

Java LinkedBlockingQueue 实现

2021-12-27 105

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 本文着重介绍 Java 并发容器中 LinkedBlockingQueue 的实现方式。

引言

本文着重介绍 Java 并发容器中 LinkedBlockingQueue 的实现方式。所有关于 Java 并发的文章均收录于<Java并发系列文章>。

LinkedBlockingQueue

LinkedBlockingQueue 底层基于单向链表实现的阻塞队列，可以当做无界队列也可以当做有界队列来使用，满足FIFO的特性，为了防止 LinkedBlockingQueue 容量迅速增，损耗大量内存。通常在创建LinkedBlockingQueue 对象时，会指定其大小，如果未指定，容量等于Integer. MAX_VALUE。那么什么是阻塞队列呢？我们知道队列有入队出队两个操作，所谓阻塞队列，就是说如果队列已满时，可以阻塞入队操作，而如果队列为空时，可以阻塞出队操作。

为了实现阻塞效果并保证线程安全，它的内部用到了两个锁和两个Condition。

/** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();

/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();

/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();

/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();

在进行数据出队时，先要获得 takeLock，然后检查当前队列容量是否为 0，如果队列容量为 0，则在 notEmpty 上等待，否则直接执行出队操作。最后判断一下，是不是执行出队操作之前，队列已经达到最大容量，如果是的话，就唤醒等待中的入队操作。

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();
        }
        // 否则，执行出队操作，并修改 size
        x = dequeue();
        c = count.getAndDecrement();
        if (c > 1)
            // 如果队列不为空，则唤醒下一个等待中的出队操作
            notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    // 如果之前队列满了，则唤醒等待中的入队操作
    if (c == capacity)
        signalNotFull();
    return x;
}

/**
 * Signals a waiting put. Called only from take/poll.
 */
private void signalNotFull() {
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        notFull.signal();
    } finally {
        putLock.unlock();
    }
}

入队操作和出队操作正好相反，同样先获取锁，不过这里用的是 putLock，检查当前队列是否已满，是的话就在 notFull 上等待，否则执行入队操作并修改size，如果之前的队列长度为 0，那么就有可能有一些出队操作被阻塞了，所以我们这里需要唤醒所有在 notEmpty 上等待的线程。

/**
 * Inserts the specified element at the tail of this queue, waiting if
 * necessary for space to become available.
 *
 * @throws InterruptedException {@inheritDoc}
 * @throws NullPointerException {@inheritDoc}
 */
public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    // Note: convention in all put/take/etc is to preset local var
    // holding count negative to indicate failure unless set.
    int c = -1;
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    // 先获取锁
    putLock.lockInterruptibly();
    try {
        /*
         * Note that count is used in wait guard even though it is
         * not protected by lock. This works because count can
         * only decrease at this point (all other puts are shut
         * out by lock), and we (or some other waiting put) are
         * signalled if it ever changes from capacity. Similarly
         * for all other uses of count in other wait guards.
         */
        while (count.get() == capacity) {
            notFull.await();
        }
        enqueue(node);
        c = count.getAndIncrement();
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        signalNotEmpty();
}

/**
 * Signals a waiting take. Called only from put/offer (which do not
 * otherwise ordinarily lock takeLock.)
 */
private void signalNotEmpty() {
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
}

LinkedBlockingQueue 除了阻塞版的入队出队操作外，当然也有不阻塞的接口，不过这些接口比较简单，基本上就是在上述基础取消 await 和 signal 逻辑，这里就不再赘述了。