LinkedBlockingQueue源码分析
2017-03-24 本文已影响126人
此鱼不得水
这几周陆续都有其他事情，项目跟进，回学校答辩等等。耽误了博客的更新力度，趁着现在有点时间空余，接着看看其他的内容。本次的主角依然是列表家族的常用成员-LinkedBlockingQueue。
LinkedBlockingQueue简单介绍
LinkedBlockingQueue作为一个链表式的阻塞队列,他与ArrayBlockingQueue的却别就好比LinkedList于ArrayList的区别一样，但是它更常用于线程池的构造函数中作为阻塞队列出现，而且为了避免队列长度的无限制增长一般需要限制阻塞列表的长度。
源码分析
类定义
/**
*都是一些常规的方法抽象
*/
public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable
属性分析
/**队列容量(重要，如果没有给出的话就默认为Integer.MAX_VALUE) */
private final int capacity;
/**队列实际长度(这里不能指定int,否则final修改会出错，但是因为是线程安全的队列，所以使用AutomicInteger)*/
private final AtomicInteger count = new AtomicInteger(0);
/**
* 头部节点(这里的transient在以前的博文中介绍过)
*/
private transient Node<E> head;
/**
* 尾部节点
*/
private transient Node<E> last;
/** take, poll等操作需要的锁 */
private final ReentrantLock takeLock = new ReentrantLock();
/** takeLock对应的Condition */
private final Condition notEmpty = takeLock.newCondition();
/** put, offer需要的锁*/
private final ReentrantLock putLock = new ReentrantLock();
/** putLock对应的Condition*/
private final Condition notFull = putLock.newCondition();
内部类
static class Node<E> {
E item;
/** 指向下一个节点的指针*/
Node<E> next;
Node(E x) { item = x; }
}
构造函数
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
/**
* 默认初始化容量为给定的值，没有给定就初始为Integer.MAX_VALUE
*/
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
last = head = new Node<E>(null);
}
/**
*将c中的数据塞到队列中，并初始化队列长度为Integer.MAX_VALUE而不是c.size();
*/
public LinkedBlockingQueue(Collection<? extends E> c) {
this(Integer.MAX_VALUE);
//这里统一说下这种写法的好处(JDK源码中处处可见)：
//1.避免直接饮用成员变量，引用局部变量的引用更加迅速。
//2.当成员变量出现被其他线程改变时，因为在内部重新将饮用定义为final，所以在方法内部的数据是一致的。
//3.让代码更短，如果成员变量(静态变量)更长的话可以定义一个短一点的名字
final ReentrantLock putLock = this.putLock;
putLock.lock();
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();
}
}
基础方法
private void enqueue(Node<E> node) {
// assert putLock.isHeldByCurrentThread(); 线程安全性需要调用者来保障
//代码从右往左读
last = last.next = node;
}
/**
* 从队列头部移除一个元素(调用者保证队列不为空)
* 返回被移除的元素的下一个元素的内容值，并且保证新的head.item=null
*/
private E dequeue() {
// assert takeLock.isHeldByCurrentThread();
线程安全性需要调用者来保障
// assert head.item == null;
Node<E> h = head;
Node<E> first = h.next;
h.next = h; // 为了加快对于h对象的清理。(还好JVM内部默认的不是引用计数法)
head = first;
E x = first.item;
first.item = null;
return x;
}
/**
* 内容简单并且重复
*/
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
notEmpty.signal();
} finally {
takeLock.unlock();
}
}
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
notFull.signal();
} finally {
putLock.unlock();
}
}
void fullyLock() {
putLock.lock();
takeLock.lock();
}
void fullyUnlock() {
takeLock.unlock();
putLock.unlock();
}
/**
*put操作会响应中断，而且e不能为Null
*/
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);
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
/*
* 因为put获得了putLock，所以在这里所有对于count的增加操作都是不可能的，只能减少count的值，如果在这里阻塞的话，一旦有其他减少count的操作就会立马被唤醒。
* 但是因为在notFull.await()的时候释放了锁，所以有可能这时候忽然有一个类put的操作提交抢占了锁，这就导致count的值又增加。所以需要在这里使用while多次判断
*/
while (count.get() == capacity) {
notFull.await();
}
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity) //如果至少还有一个坑
notFull.signal();
} finally {
putLock.unlock();
}
//如果原来列表为空，这就要通知一下
if (c == 0)
signalNotEmpty();
}
/**
*offer于put的区别就是可以设置等待锁的时间，而且有返回值代表成功与否
*/
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) {
if (nanos <= 0)//等待时间到了之后立马返回(finally释放锁)
return false;
nanos = notFull.awaitNanos(nanos);//返回的nanos代表已经等待的时间减去给定的等待时间
}
//以下部分跟put操作一样
enqueue(new Node<E>(e));
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
return true;
}
//put操作在full情况会等待，而offer直接返回失败
public boolean offer(E e) {
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);
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)
signalNotEmpty();
return c >= 0; //如果容量满了的话c=-1，说明offer操作失败
}
/**
*take操作是个阻塞操作，与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的原因跟put操作的原因一样
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;
}
//这两个操作与offer操作对应，实现也是一模一样(看过offer的可以跳过这个了)
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
E x = null;
int c = -1;
long nanos = unit.toNanos(timeout);
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() == 0) {
if (nanos <= 0)
return null;
nanos = notEmpty.awaitNanos(nanos);
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}
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;
}
//peek:偷窥(只是看看而不改变链表结构)
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();
}
}
/**
* remove操作是全锁的，因为这个操作会改变链表中间的节点。但是只会移除第一个匹配的节点(遍历时候也可以进行移除呦)
* remove和contains都是需要全锁的，保证在操作时候不会有其他线程改变现有结构(因为这个两个操作都是需要在“某个瞬间时刻”进行的)
*/
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) {
// 一定要在isFullyLocked()使用的前提下;
// 这里没有改变p.next是为了保持在遍历时候的弱一致性
p.item = null;
trail.next = p.next;
if (last == p)
last = trail;
if (count.getAndDecrement() == capacity)
notFull.signal(); //注意这里已经调用了isFullyLocked()所以已经取得了锁
}
/**
* 将链表中指定数量(maxElemets)的的内容放到c中(从头部开始)
*/
public int drainTo(Collection<? super E> c, int maxElements) {
if (c == null)
throw new NullPointerException();
if (c == this)
throw new IllegalArgumentException();
boolean signalNotFull = false;
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {//链表中的元素数量可能小于maxElements
int n = Math.min(maxElements, count.get());
Node<E> h = head;
int i = 0;
try {
while (i < n) { //从n的值保证了p不为null
Node<E> p = h.next;
c.add(p.item);
p.item = null;
h.next = h; //促进GC
h = p;
++i;
}
return n;
} finally {
// 即使在插入过程抛出异常，已经进行的操作还会保持有效
if (i > 0) {
// 正常情况下h.item = null
head = h;
//如果当前count == capacity就表明队列已经从满-》不满，这样就可以通知等待在notFull上的线程
signalNotFull = (count.getAndAdd(-i) == capacity);
}
}
} finally {
takeLock.unlock();
//因为这是一个减少队列元素的过程，所以有了元素移除操作就要看一下是否有线程等待在notFull.
if (signalNotFull)
signalNotFull();
}
}
LinkedBlockingQueue的内容都在这里进行了介绍。下面我们总结一下它与LinkedList的区别
- LinkedBlockingQueue线程安全，LinkedList不安全
- LinkedBlockingQueue为单向链表，LinkedList为双向链表
- LinkedBlockingQueue支持容量设置，并且不能扩容，LinkedList不支持设置容量
