转载

JDK源码分析-LinkedBlockingQueue

概述

前文「 JDK源码分析-ArrayBlockingQueue 」分析了 ArrayBlockingQueue 的代码实现，LinkedBlockingQueue 也是阻塞队列的实现。与前者不同的是，后者内部是由链表实现的。

LinkedBlockingQueue 的继承结构如下：

下面分析其主要方法的代码实现。

代码分析

LinkedBlockingQueue 内部有一个嵌套类 Node，它表示链表的节点，如下：

static class Node<E> {

E item; // 节点元素

Node<E> next; // 后继节点

Node(E x) { item = x; }

}

PS: 从 Node 定义可以看出该链表是一个单链表。

主要成员变量

// 链表的容量（若不指定则为 Integer.MAX_VALUE）

// 当前元素的数量（即链表中元素的数量）

private final AtomicInteger count = new AtomicInteger();

// 链表的头节点（节点元素为空）

transient Node<E> head;

// 链表的尾结点（节点元素为空）

// take、poll 等出队操作持有的锁

private final ReentrantLock takeLock = new ReentrantLock();

/** Wait queue for waiting takes */

// 出队锁的条件队列

// put、offer 等入队操作的锁

private final ReentrantLock putLock = new ReentrantLock();

/** Wait queue for waiting puts */

// 入队锁的条件队列

构造器

LinkedBlockingQueue 有三个构造器，分别如下：

// 构造器 1：无参构造器，初始容量为 Integer.MAX_VALUE，即 2^31-1

}

// 构造器 2：指定容量的构造器

public LinkedBlockingQueue(int capacity) {

if (capacity <= 0) throw new IllegalArgumentException();

// 初始化链表的头尾节点

last = head = new Node<E>(null);

}

// 构造器 3：用给定集合初始化的构造器

// 调用构造器 2 进行初始化

final ReentrantLock putLock = this.putLock;

putLock.lock(); // Never contended, but necessary for visibility

throw new IllegalStateException("Queue full");

// 将集合中的元素封装成 Node 对象，并添加到链表末尾

enqueue(new Node<E>(e));

++n;

}

count.set(n);

} finally {

putLock.unlock();

}

}

enqueue 方法如下：

// 将 node 节点添加到链表末尾

private void enqueue(Node<E> node) {

last = last.next = node;

}

主要入队方法：put(E)、 offer(E, timeout, TimeUnit) 、offer(E)

1. put(E) 代码如下：

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.

// 把 E 封装成 Node 节点

Node<E> node = new Node<E>(e);

putLock.lockInterruptibly();

/*

* Note that count is used in wait guard even though it is

* not protected by lock. This works because count can

*> * 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.

*/

// 若队列已满，notFull 等待（类比生产者）

notFull.await();

}

// node 入队

enqueue(node);

c = count.getAndIncrement();

// 若该元素添加后，队列仍未满，唤醒一个其他生产者线程

notFull.signal();

putLock.unlock();

}

// c==0 说明之前队列为空，出队线程处于等待状态，

// 添加一个元素后，将出队线程唤醒（消费者）

signalNotEmpty();

}

signalNotEmpty 方法：

final ReentrantLock takeLock = this.takeLock;

takeLock.lock();

// 唤醒 notEmpty 条件下的一个线程（消费者）

notEmpty.signal();

takeLock.unlock();

}

}

2. offer(E, timeout, TimeUnit):

<span style=";margin: 0px; padding: 0px; max-width: 1000%; box-sizing: border-box !important; overflow-wrap: break-word !important;;margin: 0px; padding: 0px; max-width: 1000%; box-sizing: border-box !important; overflow-wrap: break-word !important;">public boolean offer(E e, long timeout, TimeUnit unit)

if (e == null) throw new NullPointerException();

putLock.lockInterruptibly();

// 等待超时，返回 false

nanos = notFull.awaitNanos(nanos);

}

// 入队

c = count.getAndIncrement();

notFull.signal();

putLock.unlock();

}

signalNotEmpty();

}

该方法与 put 操作类似，不同的是 put 方法在队列满时会一直等待，而该方法有超时时间，超时后返回 false。

3. offer(E):

if (e == null) throw new NullPointerException();

final AtomicInteger count = this.count;

// 若队列已满，直接返回 false

if (count.get() == capacity)

int c = -1;

Node<E> node = new Node<E>(e);

putLock.lock();

// 队列未满，入队

if (count.get() < capacity) {

enqueue(node);

if (c + 1 < capacity)

// 队列未满，唤醒 notFull 下的线程，继续入队

notFull.signal();

}

} finally {

putLock.unlock();

}

if (c == 0)

signalNotEmpty();

return c >= 0;

}

入队方法小结

1. put(E): 若队列已满，则等待，无返回值；

2. offer(E, timeout, TimeUnit): 与 put 方法类似，有超时等待和返回值；

3. offer(E): 立即返回，没有循环等待。

常用出队方法：take、poll(timeout, TimeUnit)、poll()、peek()

1. take():

public E take() throws InterruptedException {

final AtomicInteger count = this.count;

takeLock.lockInterruptibly();

// 队列为空，notEmpty 条件下的线程等待（消费者）

while (count.get() == 0) {

notEmpty.await();

}

// 从队列头部删除节点

x = dequeue();

// 若队列不为空，唤醒一个 notEmpty 条件下的线程（消费者）

if (c > 1)

notEmpty.signal();

} finally {

takeLock.unlock();

}

// 队列已经不满了，唤醒 notFull 条件下的线程（生产者）

signalNotFull();

}

dequeue 方法：

Node<E> h = head; // 头节点

Node<E> first = h.next; // 头节点的后继节点

h.next = h; // help GC // 后继节点指向自己（从链表中删除）

head = first; // 更新头节点

E x = first.item; // 获取要删除节点的数据

first.item = null; // 清空数据（新的头节点）

}

signalNotFull:

// 唤醒生产者

notFull.signal();

putLock.unlock();

}

}

2. poll(timeout, TimeUnit) :

public E poll(long timeout, TimeUnit unit) throws InterruptedException {

E x = null;

takeLock.lockInterruptibly();

// 队列已空

// 超时返回 null

nanos = notEmpty.awaitNanos(nanos);

}

x = dequeue();

// 若队列不空，唤醒一个 notEmpty 条件下的线程（消费者）

c = count.getAndDecrement();

notEmpty.signal();

takeLock.unlock();

}

// 队列不满，唤醒 notFull 条件下的线程（生产者）

signalNotFull();

}

与 take 方法类似，多了超时等待。

3. poll():

final AtomicInteger count = this.count;

// 队列为空，返回 null

if (count.get() == 0)

takeLock.lock();

// 队列不为空，出队

if (count.get() > 0) {

x = dequeue();

// 该元素出队后，队列仍不为空，唤醒其他消费者

if (c > 1)

notEmpty.signal();

}

} finally {

takeLock.unlock();

}

// 队列已经不满，唤醒生产者

signalNotFull();

}

4. peek()

if (count.get() == 0)

takeLock.lock();

// 头节点的后继节点

Node<E> first = head.next;

else

takeLock.unlock();

}

}

peek() 方法只返回头节点，并不删除。严格来说该方法并不属于出队操作，只是查询。

出队方法小结

1. take(): 获取队列头部元素，并将其移除，队列为空时阻塞等待；

2. poll(long, unit): 获取队列头部元素，并将其移除，队列为空时等待一段时间，若超时返回 null；

3 . poll() : 获取队列头部元素，并将其移除，队列为空时返回 null；

小结

1. LinkedBlockingQueue 是基于单链表的阻塞队列实现，它在初始化时可以指定容量，若未指定，则默认容量为 Integer.MAX_VALUE；

2. 内部使用了 ReentrantLock 保证线程安全；

3. 常用方法：

入队：put, offer

出队：take, poll, peek

LinkedBlockingQueue 与 ArrayBlockingQueue 比较：

1. ArrayBlockingQueue 使用单个锁，可以指定是否公平；而 LinkedBlockingQueue 内部使用了两个锁：putLock 和 takeLock，都是非公平锁。

2. 入队出队区别

入队时， LinkedBlockingQueue 会判断当前元素入队后，队列是否已满，若未满，则唤醒其他生产者线程；而入队后，当队列之前为空时才唤醒其他消费者线程。ArrayBlockingQueue 则是每次入队都会唤醒消费者线程。

出队时， LinkedBlockingQueue 会判断当前元素出队后，队列是否已空，若未空，则唤醒其他消费者线程；而出队后，当队列之前为满时才唤醒其他生产者线程。 ArrayBlockingQueue 则是每次出队都会唤醒生产者线程。

Stay hungry, stay foolish.

JDK源码分析-LinkedBlockingQueue

原文 https://juejin.im/post/5d614b0a518825121f66187c

正文到此结束

所属分类：编程技术 Java

本文标签： value Atom https 时间 queue db 代码 Word node 数据删除源码 UI http 安全 src id cat final Collection 线程锁 IO
版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
本文海报： 生成海报一生成海报二

其他链接

关于本站

本站定位：个人技术类博客

本站作用：写博客、记日志、闲聊扯淡鼓捣技术。

问题交流

JDK源码分析-LinkedBlockingQueue

热门推荐

相关文章

说给你听

本文目录

随机标签

书籍教程

近期评论

网站信息

其他链接

关于本站

问题交流