概览
1. 基于链表的可选有界阻塞队列。根据FIFO的出入队顺序,从队列头部检索和获取元素,在队列尾部插入新元素。
2. 当作为有界阻塞队列,在队列空间不足时,put方法将会一直阻塞直到有多余空间才会执行插入元素操作,take方法则相反,只到队列内元素不为空时,才将队列元素逐个取出。
3. 队列容量不指定时,默认为Integer.MAX_VALUE,此时可以看作无界队列。
4. 使用非公平锁进行并发控制。所有方法都是线程安全的。
使用方法
下面的文章给出了阻塞队列的四种基本用法:
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 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 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 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 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 item; Node 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 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 h = head; Node 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属性需要注意下,这里使用了原子类保证操作的原子性。后面的入队和出队,将会频繁使用它。
/** 容量, 初始化不设置时默认为Integer.MAX_VALUE*/ private final int capacity; /** 当前队列内的元素数量 */ private final AtomicInteger count = new AtomicInteger(); /** * 队首 * 不变量: head.item == null */ transient Node head; /** * 队尾 * 不变量: last.next == null */ private transient Node 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 (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)); ++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 node = new Node(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
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 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; } // 唤醒生产者 /** * 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(); } }
总结
总体上看LinkedBlockingQueue类不难,整个生产-消费的流程实现也比较简单。源码已经把该介绍的东西都讲得很明白了,我这属于依葫芦画瓢顺着源码注释写出来的。这么一写,自己这个类的印象就很深刻了。
以上为个人经验,希望能给大家一个参考,也希望大家多多支持IT俱乐部。