LinkedBlockingQueue 源码学习

本文主要是介绍LinkedBlockingQueue 源码学习，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1. 概况

FIFO
初始化时候指定了链表长度，默认为 Integer.MAX_VALUE，而节点是插入时才创建

2. 类定义

继承父类与实现接口与 ArrayBlockingQueue 一样。

public class LinkedBlockingQueue<E> extends AbstractQueue<E>implements BlockingQueue<E>, java.io.Serializable {
}

3. 成员变量

与 ArrayBlockingQueue 不一样的地方：

使用两个 Lock 控制入队和出队；ArrayBlockingQueue 只有一个 LOCK 控制并发访问
使用 AtomicInteger 记录当前链表的元素个数；ArrayBlockingQueue 则是使用一个基本类型的变量。

/** 容量大小，默认为 Integer.MAX_VALUE */
private final int capacity;/** 链表中元素个数 */
private final AtomicInteger count = new AtomicInteger();
/*** 链表的头节点，这个头节点是空节点，即 head.item = null*/
transient Node<E> head;
/*** 链表的尾节点，也是一个空节点，即 last.item = null*/
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();

4. 构造方法

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 visibilitytry {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();}
}// 相当于 last.next = node; last = last.next
private void enqueue(Node<E> node) {// assert putLock.isHeldByCurrentThread();// assert last.next == null;last = last.next = node;
}

5. 成员方法

5.1 入队方法

5.1.1 put

使用 put 锁控制其它线程进行相关入队操作
局部变量 c 用于唤醒正在等待的出队方法

public void put(E e) throws InterruptedException {if (e == null) throw new NullPointerException();int c = -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) {notFull.await();}// 入队操作enqueue(node);// getAndIncrement 方法的返回是自增的元素，所以 c 被赋值添加元素之前的链表节点个数；并且 count++c = count.getAndIncrement();// 添加一个元素还没有满，通知等待队列，还可以进行入队操作if (c + 1 < capacity)notFull.signal();} finally {putLock.unlock();}// 若入队前队列size=0，此时队列元素个数为1，那么就唤醒正在等待出队的方法if (c == 0)signalNotEmpty(); 
}

5.1.2 offer

与 put 方法类似，都是对 putLock 进行加锁控制
带超时时间的 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>(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;
}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)return false;nanos = notFull.awaitNanos(nanos);}enqueue(new Node<E>(e));c = count.getAndIncrement();if (c + 1 < capacity)notFull.signal();} finally {putLock.unlock();}if (c == 0)signalNotEmpty();return true;
}

5.2 出队方法

5.2.1 poll

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();  // 先返回再减 1if (c > 1)notEmpty.signal(); // 通知其它正在等待的 出队方法}} finally {takeLock.unlock();}if (c == capacity)  // 出队之前队列是满的，现在已经出队了，那么通知其它正在等待的 入队方法signalNotFull();return x;
}// 仅在同步方法块中调用
private E dequeue() {// assert takeLock.isHeldByCurrentThread();// assert head.item == null;Node<E> h = head;Node<E> first = h.next;h.next = h; // help GChead = first;E x = first.item;first.item = null;return x;
}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;
}

5.2.2 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;elsereturn first.item;} finally {takeLock.unlock();}
}

5.2.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) { // 队列为空，进行等待notEmpty.await();}x = dequeue(); // 出队c = count.getAndDecrement();if (c > 1)notEmpty.signal();} finally {takeLock.unlock();}if (c == capacity)signalNotFull();  // 通知正在等待 入队的线程，可以放入元素了return x;
}

删除元素

从链表中删除指定元素，删除成功返回 true

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();}
}// 将节点 P 从链表中移除，trail 是 p 节点的前一个节点
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.p.item = null;trail.next = p.next;if (last == p)last = trail; // 将 last 节点置为 p 的前置节点if (count.getAndDecrement() == capacity)notFull.signal();
}