正文
【JUC】JDK1.8源码分析之LinkedBlockingQueue（四）
提示：扫一扫查出行【扫一扫了解最新限行尾号】
复制提示
一、前言
分析完了ArrayBlockingQueue后，接着分析LinkedBlockingQueue，与ArrayBlockingQueue不相同，LinkedBlockingQueue底层采用的是链表结构，其源码也相对比较简单，下面进行正式的分析。
二、LinkedBlockingQueue数据结构
从LinkedBlockingQueue的命名就大致知道其数据结构采用的是链表结构，通过源码也可以验证我们的猜测，其数据结构如下。
说明：可以看到LinkedBlockingQueue采用的是单链表结构，包含了头结点和尾节点。
三、LinkedBlockingQueue源码分析
3.1 类的继承关系
public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {}
说明：LinkedBlockingQueue继承了AbstractQueue抽象类，AbstractQueue定义了对队列的基本操作；同时实现了BlockingQueue接口，BlockingQueue表示阻塞型的队列，其对队列的操作可能会抛出异常；同时也实现了Searializable接口，表示可以被序列化。
3.2 类的内部类
LinkedBlockingQueue内部有一个Node类，表示结点，用于存放元素，其源码如下。
static class Node<E> {
// 元素
E item;
// next域
Node<E> next;
// 构造函数
Node(E x) { item = x; }
}
说明：Node类非常简单，包含了两个域，分别用于存放元素和指示下一个结点。
3.3 类的属性
public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
// 版本序列号
private static final long serialVersionUID = -6903933977591709194L;
// 容量
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();
}
说明：可以看到LinkedBlockingQueue包含了读、写重入锁（与ArrayBlockingQueue不同，ArrayBlockingQueue只包含了一把重入锁），读写操作进行了分离，并且不同的锁有不同的Condition条件（与ArrayBlockingQueue不同，ArrayBlockingQueue是一把重入锁的两个条件）。
3.4 类的构造函数
1. LinkedBlockingQueue()型构造函数
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
说明：该构造函数用于创建一个容量为 Integer.MAX_VALUE 的 LinkedBlockingQueue。
2. LinkedBlockingQueue(int)型构造函数
public LinkedBlockingQueue(int capacity) {
// 初始化容量必须大于0
if (capacity <= 0) throw new IllegalArgumentException();
// 初始化容量
this.capacity = capacity;
// 初始化头结点和尾结点
last = head = new Node<E>(null);
}
说明：该构造函数用于创建一个具有给定（固定）容量的 LinkedBlockingQueue。
3. LinkedBlockingQueue(Collection<? extends E>)型构造函数
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) { // 遍历c集合
if (e == null) // 元素为null,抛出异常
throw new NullPointerException();
if (n == capacity) //
throw new IllegalStateException("Queue full");
enqueue(new Node<E>(e));
++n;
}
count.set(n);
} finally {
putLock.unlock();
}
}
说明：该构造函数用于创建一个容量是 Integer.MAX_VALUE 的 LinkedBlockingQueue，最初包含给定 collection 的元素，元素按该 collection 迭代器的遍历顺序添加。
3.5 核心函数分析
1. put函数
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条件上进行等待
notFull.await();
}
// 入队列
enqueue(node);
// 更新元素个数，返回的是以前的元素个数
c = count.getAndIncrement();
if (c + 1 < capacity) // 元素个数是否小于容量
// 唤醒在notFull条件上等待的某个线程
notFull.signal();
} finally {
// 释放锁
putLock.unlock();
}
if (c == 0) // 元素个数为0，表示已有take线程在notEmpty条件上进入了等待，则需要唤醒在notEmpty条件上等待的线程
signalNotEmpty();
}
说明：put函数用于存放元素，其流程如下。
① 判断元素是否为null，若是，则抛出异常，否则，进入步骤②
② 获取存元素锁，并上锁，如果当前线程被中断，则抛出异常，否则，进入步骤③
③ 判断当前队列中的元素个数是否已经达到指定容量，若是，则在notFull条件上进行等待，否则，进入步骤④
④ 将新生结点入队列，更新队列元素个数，若元素个数小于指定容量，则唤醒在notFull条件上等待的线程，表示可以继续存放元素。进入步骤⑤
⑤ 释放锁，判断结点入队列之前的元素个数是否为0，若是，则唤醒在notEmpty条件上等待的线程（表示队列中没有元素，取元素线程被阻塞了）。
put函数中会调用到enqueue函数和signalNotEmpty函数，enqueue函数源码如下
private void enqueue(Node<E> node) {
// assert putLock.isHeldByCurrentThread();
// assert last.next == null;
// 更新尾结点域
last = last.next = node;
}
说明：可以看到，enqueue函数只是更新了尾节点。signalNotEmpty函数源码如下
private void signalNotEmpty() {
// 取元素锁
final ReentrantLock takeLock = this.takeLock;
// 获取锁
takeLock.lock();
try {
// 唤醒在notEmpty条件上等待的某个线程
notEmpty.signal();
} finally {
// 释放锁
takeLock.unlock();
}
}
说明：signalNotEmpty函数用于唤醒在notEmpty条件上等待的线程，其首先获取取元素锁，然后上锁，然后唤醒在notEmpty条件上等待的线程，最后释放取元素锁。
2. offer函数
public boolean offer(E e) {
// 确保元素不为null
if (e == null) throw new NullPointerException();
// 获取计数器
final AtomicInteger count = this.count;
if (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条件上等待的某个线程
notFull.signal();
}
} finally {
// 释放锁
putLock.unlock();
}
if (c == 0) // 元素个数为0，则唤醒在notEmpty条件上等待的某个线程
signalNotEmpty();
return c >= 0;
}
说明：offer函数也用于存放元素，offer函数添加元素不会抛出异常（其他的域put函数类似）。
3. take函数
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) { // 元素个数为0
// 在notEmpty条件上等待
notEmpty.await();
}
// 出队列
x = dequeue();
// 更新元素个数，返回的是以前的元素个数
c = count.getAndDecrement();
if (c > 1) // 元素个数大于1，则唤醒在notEmpty上等待的某个线程
notEmpty.signal();
} finally {
// 释放锁
takeLock.unlock();
}
if (c == capacity) // 元素个数到达指定容量
// 唤醒在notFull条件上等待的某个线程
signalNotFull();
// 返回
return x;
}
说明：take函数用于获取一个元素，其与put函数相对应，其流程如下。
① 获取取元素锁，并上锁，如果当前线程被中断，则抛出异常，否则，进入步骤②
② 判断当前队列中的元素个数是否为0，若是，则在notEmpty条件上进行等待，否则，进入步骤③
③ 出队列，更新队列元素个数，若元素个数大于1，则唤醒在notEmpty条件上等待的线程，表示可以继续取元素。进入步骤④
④ 释放锁，判断结点出队列之前的元素个数是否为指定容量，若是，则唤醒在notFull条件上等待的线程（表示队列已满，存元素线程被阻塞了）。
take函数调用到了dequeue函数和signalNotFull函数，dequeue函数源码如下
private E dequeue() {
// assert takeLock.isHeldByCurrentThread();
// assert head.item == null;
// 头结点
Node<E> h = head;
// 第一个结点
Node<E> first = h.next;
// 头结点的next域为自身
h.next = h; // help GC
// 更新头结点
head = first;
// 返回头结点的元素
E x = first.item;
// 头结点的item域赋值为null
first.item = null;
// 返回结点元素
return x;
}
说明：dequeue函数的作用是将头结点更新为之前头结点的下一个结点，并且将更新后的头结点的item域设置为null。signalNotFull函数的源码如下
private void signalNotFull() {
// 存元素锁
final ReentrantLock putLock = this.putLock;
// 获取锁
putLock.lock();
try {
// 唤醒在notFull条件上等待的某个线程
notFull.signal();
} finally {
// 释放锁
putLock.unlock();
}
}
说明：signalNotFull函数用于唤醒在notFull条件上等待的某个线程，其首先获取存元素锁，然后上锁，然后唤醒在notFull条件上等待的线程，最后释放存元素锁。
4. poll函数
public E poll() {
// 获取计数器
final AtomicInteger count = this.count;
if (count.get() == 0) // 元素个数为0
return null;
//
E x = null;
int c = -1;
// 取元素锁
final ReentrantLock takeLock = this.takeLock;
// 获取锁
takeLock.lock();
try {
if (count.get() > 0) { // 元素个数大于0
// 出队列
x = dequeue();
// 更新元素个数，返回的是以前的元素个数
c = count.getAndDecrement();
if (c > 1) // 元素个数大于1
// 唤醒在notEmpty条件上等待的某个线程
notEmpty.signal();
}
} finally {
// 释放锁
takeLock.unlock();
}
if (c == capacity) // 元素大小达到指定容量
// 唤醒在notFull条件上等待的某个线程
signalNotFull();
// 返回元素
return x;
}
说明：poll函数也用于存放元素，poll函数添加元素不会抛出异常（其他的与take函数类似）。
5. remove函数
public boolean remove(Object o) {
// 元素为null，返回false
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();
}
}
说明：remove函数的流程如下
① 获取读、写锁（防止此时继续出、入队列）。进入步骤②
② 遍历链表，寻找指定元素，若找到，则将该结点从链表中断开，有利于被GC，进入步骤③
③ 释放读、写锁（可以继续出、入队列）。步骤②中找到指定元素则返回true，否则，返回false。
其中，remove函数会调用unlink函数，其源码如下
void unlink(Node<E> p, Node<E> trail) {
// assert isFullyLocked();
// p.next is not changed, to allow iterators that are
// traversing p to maintain their weak-consistency guarantee.
// 结点的item域赋值为null
p.item = null;
// 断开p结点
trail.next = p.next;
if (last == p) // 尾节点为p结点
// 重新赋值尾节点
last = trail;
if (count.getAndDecrement() == capacity) // 更新元素个数，返回的是以前的元素个数，若结点个数到达指定容量
// 唤醒在notFull条件上等待的某个线程
notFull.signal();
}
说明：unlink函数用于将指定结点从链表中断开，并且更新队列元素个数，并且判断若之前队列元素的个数达到了指定容量，则会唤醒在notFull条件上等待的某个线程。
四、示例
下面通过一个示例来了解LinkedBlockingQueue的使用。
package com.hust.grid.leesf.collections; import java.util.concurrent.LinkedBlockingQueue;
class PutThread extends Thread {
private LinkedBlockingQueue<Integer> lbq;
public PutThread(LinkedBlockingQueue<Integer> lbq) {
this.lbq = lbq;
} public void run() {
for (int i = 0; i < 10; i++) {
try {
System.out.println("put " + i);
lbq.put(i);
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
} class GetThread extends Thread {
private LinkedBlockingQueue<Integer> lbq;
public GetThread(LinkedBlockingQueue<Integer> lbq) {
this.lbq = lbq;
} public void run() {
for (int i = 0; i < 10; i++) {
try {
System.out.println("take " + lbq.take());
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public class LinkedBlockingQueueDemo {
public static void main(String[] args) {
LinkedBlockingQueue<Integer> lbq = new LinkedBlockingQueue<Integer>(); PutThread p1 = new PutThread(lbq);
GetThread g1 = new GetThread(lbq); p1.start();
g1.start();
}
}
运行结果：
put 0
take 0
put 1
take 1
put 2
take 2
put 3
take 3
put 4
take 4
put 5
take 5
put 6
take 6
put 7
take 7
put 8
take 8
put 9
take 9
说明：示例中使用了两个线程，一个用于存元素，一个用于读元素，存和读各10次，每个线程存一个元素或者读一个元素后都会休眠100ms，可以看到结果是交替打 印，并且首先打印的肯定是put线程语句（因为若取线程先取元素，此时队列并没有元素，其会阻塞，等待存线程存入元素），并且最终程序可以正常结束。
① 若修改取元素线程，将存的元素的次数修改为15次（for循环的结束条件改为15即可），运行结果如下：
put 0
take 0
put 1
take 1
put 2
take 2
put 3
take 3
put 4
take 4
put 5
take 5
put 6
take 6
put 7
take 7
put 8
take 8
put 9
take 9
说明：运行结果与上面的运行结果相同，但是，此时程序无法正常结束，因为take方法被阻塞了，等待被唤醒。
五、总结
LinkedBlockingQueue的源码相对比较简单，其也是通过ReentrantLock和Condition条件来保证多线程的正确访问的，并且取元素（出队列）和存元素（入队列）是采用不同的锁，进行了读写分离，有利于提高并发度。LinkedBockingQueue的分析就到这里，欢迎交流，谢谢各位园友的观看~
