LinkedBlockingQueue链式阻塞队列的使用和原理解析

2022-11-13 19:11:39 队列链式阻塞

概览

1. 基于链表的可选有界阻塞队列。根据FIFO的出入队顺序，从队列头部检索和获取元素，在队列尾部插入新元素。

2. 当作为有界阻塞队列，在队列空间不足时，put方法将会一直阻塞直到有多余空间才会执行插入元素操作，take方法则相反，只到队列内元素不为空时，才将队列元素逐个取出。

3. 队列容量不指定时，默认为Integer.MAX_VALUE，此时可以看作无界队列。

4. 使用非公平锁进行并发控制。所有方法都是线程安全的。

使用方法

下面的文章给出了阻塞队列的四种基本用法：

了解BlockingQueue 大体框架和实现思路

LinkedBlockingQueue实现了BlockingQueue类。

在BlockingQueue中，方法被分为如下四类：

Throws exception：操作未实现时（正常流程下的执行）抛出异常
Special value：根据操作的实际情况，返回特定值，例如null、false（这些失败可能是线程中断、队列为空引起的）
Blocks：阻塞当前线程，直到当前线程可以成功执行
Times out：尝试指定时间后，放弃执行

	Throws exception	Special value	Blocks	Times out
新增	add(E e)	offer(E e)	put(E e)	offer(E e, long timeout, TimeUnit unit)
删除	remove()	poll()	take()	poll(long timeout, TimeUnit unit)
查询	element()	peek()

1. add | remove | element

这三个方法在BlockingQueue的定义中，都会在操作未实现时，抛出异常。

add(E e)：在队尾添加元素e，add内部调用offer方法实现。因此，元素e为空时，抛出NullPointerException异常；插入失败时，抛出IllegalStateException异常。
remove：删除队首元素，内部调用poll方法。队首无数据时，抛出NoSuchElementException异常。
element：检索队首元素。队首无数据时，抛出NoSuchElementException异常。

LinkedBlockingQueue<Integer> blockingQue = new LinkedBlockingQueue<>();
blockingQue.add(1);
blockingQue.remove(1);
blockingQue.remove(); // NoSuchElementException
blockingQue.element(); // NoSuchElementException

2. offer | poll | peek

根据操作的实际情况，返回特定值，例如null、false（这些失败可能是线程中断、队列为空引起的）

offer(E e)：在队尾添加元素e，元素e为空时，抛出NullPointerException异常；插入失败时返回false。
poll：删除队首元素。删除失败时返回false。
peek：检索队首元素。队首无数据时，返回null。

LinkedBlockingQueue<Integer> blockingQue = new LinkedBlockingQueue<>();
blockingQue.offer(1);
blockingQue.poll();
Integer peek = blockingQue.peek(); // 返回null

3. put | take

阻塞当前线程，直到当前线程可以成功执行。

put(E e)：在队尾添加元素e，元素e为空时，抛出NullPointerException异常。当队列满时，阻塞put线程，等待队列被消费后，队列容量不满时，该阻塞线程继续尝试在队尾插入元素。该方法在阻塞时可以被中断，并抛出InterruptedException异常。
take：删除并获取队首元素。队首元素不为空时返回。队首元素为空，阻塞take线程，等待队列不为空时，再次尝试消费队首元素。该方法在阻塞时可以被中断，并抛出InterruptedException异常。

注意：阻塞时，不会解除锁占用。

LinkedBlockingQueue<Integer> blockingQue = new LinkedBlockingQueue<>();
try {
    blockingQue.put(1);
    blockingQue.take();
} catch (InterruptedException e) {
    // 线程被中断
    e.printStackTrace();
}

4. offer | poll （timeout）

尝试指定时间后，放弃执行

offer(E e, long timeout, TimeUnit unit)：在队尾添加元素e，元素e为空时，抛出NullPointerException异常；当队列容量满时，线程休眠一定时间后再次查看队列容量，当该休眠时间大于等于timeout后，此时队列还满则返回false。不满时，尝试入队。需要注意的是，由于伪唤醒机制的存在，线程可能在timeout这个时间段内的任意一点被唤醒，如果队列容易不满，则会直接执行入队操作。阻塞时，当前线程被中断抛出InterruptedException异常。
poll(long timeout, TimeUnit unit)：删除队首元素。poll与offer对应的，当队列为空的时候，线程休眠一定时间。休眠时，当前线程被中断抛出InterruptedException异常。

LinkedBlockingQueue<Integer> blockingQue = new LinkedBlockingQueue<>();
try {
    blockingQue.offer(1, 100, TimeUnit.MILLISECONDS);
    blockingQue.poll(100, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
    e.printStackTrace();
}

当然除了上述的阻塞队列的基本操作外，LinkedBlockingQueue还具有集合Collection的性质。因此集合中的通用方法也可以使用。

源码解析

说明

本次源码分析主要按照下面几个步骤进行：

1. 保存队列数据的容器以及出入队方法

2. 主要成员变量以及作用

3. 主要方法分析

队列容器

结构图

仅有数据item和后继next的单向节点，结构简单。

static class node<E> {
    E item;
 
    Node<E> next;
 
    Node(E x) { item = x; }
}

next三种情况

A 普通节点的真实后继

B 真正的队首节点，item=null（队首节点恒为head.next）

C 队尾节点，next=null

item: null -> first -> …… -> last
next: first -> second -> ……-> null

入队操作

    // 入队
    private void enqueue(Node<E> node) {
        // assert putLock.isHeldByCurrentThread();
        // assert last.next == null;
        last = last.next = node; // last.next = node; last = node;
    }

步骤1

步骤2

出队操作

 
    // 出队
    private E dequeue() {
        // assert takeLock.isHeldByCurrentThread();
        // assert head.item == null;
        Node<E> h = head;
        Node<E> first = h.next;
        // 形成引用链闭环，JVM根据可达性分析时，GC root的引用链与该对象之间不可达，进行GC
        h.next = h; // help GC
        head = first;
        E x = first.item;
        first.item = null;
        return x;
    }

步骤1

步骤2

关键成员变量

队列入队和出队锁分离，都使用了非公平锁。

这里的count属性需要注意下，这里使用了原子类保证操作的原子性。后面的入队和出队，将会频繁使用它。


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

初始化

三个构造函数

    public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }
 
    // 自定义容量
    public LinkedBlockingQueue(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node<E>(null);
    }
 
    // 初始化时，批量添加集合中的元素
    public LinkedBlockingQueue(Collection<? extends E> c) {
        this(Integer.MAX_VALUE);
        final ReentrantLock putLock = this.putLock;
        putLock.lock(); // Never contended, but necessary for visibility
        try {
            int n = 0;
            for (E e : c) {
                if (e == null)
                    throw new NullPointerException();
                if (n == capacity)
                    throw new IllegalStateException("Queue full");
                enqueue(new Node<E>(e));
                ++n;
            }
            count.set(n);
        } finally {
            putLock.unlock();
        }
    }

put方法

put方法几个关注点

释放锁的时机
执行入队操作
唤醒生产者的时机
唤醒消费者的时机

这几点是整个阻塞操作的核心，可以在下面的分析中仔细观察。

注：由于阻塞队列就是基于生产者-消费者模型的，因此，下文中都把调用put方法的线程称为生产者，调用take方法的线程称为消费者。

总体分析

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; // -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) {
            // 这里是线程在执行put操作时唯一一个执行过程中释放锁的地方
            notFull.await(); // 容量已满，等待被消费后唤醒
        }
        // 添加元素，更新容量
        enqueue(node);
        c = count.getAndIncrement();
        // 队列容量有余时，在这里再次唤醒一个其他的生产者线程（或者说消费者消费速度大于生产）
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        // 释放锁
        putLock.unlock();
    }
    // 唤醒一个消费者
    if (c == 0)
        signalNotEmpty();
}

count属性并发问题

这里需要重点关注count，由于有两把锁，count可以同时被putLock、takeLock操作，那么这里是否会产生并发问题。

分析如下：

A. 只有putLock或takeLock一把锁操作：就是单线程操作，没影响，不产生并发问题。

其他所有put操作都处于await的状态或者竞争锁状态，其他线程也因为获取不到锁而无法执行，只有等该节点添加完成释放锁，其他线程才有机会继续执行。

        while (count.get() == capacity) {
            notFull.await(); // 容量已满，等待被消费后唤醒
        }

B. putLock和takeLock同时操作：我们假设两个线程一个获取到putLock，一个获取到了takeLock（同时最多也只有两个线程操作count）。

// put            
while (count.get() == capacity) {
     notFull.await();
}
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
 
 
 
// take
while (count.get() == 0) {
    notEmpty.await();
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
    notEmpty.signal();

由于count是原子类那么count的所有读写操作必然是一个串联的操作，而非并行操作，因此也不存在并发问题，如下图（顺序可能不同）：

唤醒消费者

代码的最后一段，会有唤醒一个消费者的操作。

// 唤醒一个等待中的消费者
private void signalNotEmpty() {
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
}

刚开始看到的时候很疑惑，为什么是c == 0才唤醒。如果生产者入队成功，那么c应该为如下值：

c = count.getAndIncrement();

后面看了一下count.getAndIncrement()方法定义才发现自己记混了，count.getAndIncrement()是一个原子操作，且返回值的是操作前的值。

ok，现在没问题了。

count >= 0，也就是说，只有在生产者入队前队列为空，入队成功之后才会唤醒一个消费者消费。

take方法

take方法与put方法大致相似，只是与put做相反操作。

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;
}
 
// 唤醒生产者

private void signalNotFull() {
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        notFull.signal();
    } finally {
        putLock.unlock();
    }
}

总结

总体上看LinkedBlockingQueue类不难，整个生产-消费的流程实现也比较简单。源码已经把该介绍的东西都讲得很明白了，我这属于依葫芦画瓢顺着源码注释写出来的。这么一写，自己这个类的印象就很深刻了。

以上为个人经验，希望能给大家一个参考，也希望大家多多支持。

相关文章