👋hi，我不是一名外包公司的员工，也不会偷吃茶水间的零食，我的梦想是能写高端CRUD
🔥 2025本人正在沉淀中... 博客更新速度++
👍 欢迎点赞、收藏、关注，跟上我的更新节奏
📚欢迎订阅专栏，专栏名《在2B工作中寻求并发是否搞错了什么》
前言
有了前文对简单实用的学习
【Java并发】【LinkedBlockingQueue】适合初学体质的LinkedBlockingQueue入门
聪明的你，一定会想知道更多。哈哈哈哈哈😄，下面主播就开始源码了！！！
LinkedBlockingQueue基本结构分析
既然是数组，那就不能少了Node，我们会发现LinkedBlockingQueue就是一个简单的单链表结构。
static class Node<E> {
E item;
/**
* next值
* - 正常情况，指向下一个节点
* - 为自己，表示该节点已经被逻辑删除
* - 为null，说明这是一个尾节点
*/
Node<E> next;
Node(E x) {
item = x; }
}
为了大家对源码的阅读，主播又根据自己的绘画天赋，画出了LinkedBlockingQueue的大概结构，如下图所示：
来来来一个个解释：
// 最大的Node数量限制，默认为Integer.MAX_VALUE
private final int capacity;
// 当前Node数量
private final AtomicInteger count = new AtomicInteger();
// 头节点（item为null）
transient Node<E> head;
// 尾节点（next为null）
private transient Node<E> last;
// 消费者锁，在take, poll等方法中有被使用到
private final ReentrantLock takeLock = new ReentrantLock();
// 队列不为空信号
private final Condition notEmpty = takeLock.newCondition();
// 生产者锁，在put, offer等方法中有被使用到
private final ReentrantLock putLock = new ReentrantLock();
// 队列满了信号
private final Condition notFull = putLock.newCondition();
构造方法源码分享
无参数构造，默认Node限制数量为Integer.MAX_VALUE：
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
指定最大Node数量多构造方法：
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
// 生成头尾节点
last = head = new Node<E>(null);
}
带初始集合的构造方法，和ArrayBlockingQueue也差不多。
public LinkedBlockingQueue(Collection<? extends E> c) {
this(Integer.MAX_VALUE); // 最大Node数设置为Integer.MAX_VALUE
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();
}
}
生产者方法源码
put方法
put会向队列中添加元素，但是是阻塞线程的：
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); // 放入阻塞队列
c = count.getAndIncrement(); // 当前Node数量++
// 新增Node后，当前队列还没满，告诉生产者，可以继续生产
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
// 原队列为空（这个c是加入新Node前的数量），唤醒消费者
if (c == 0)
signalNotEmpty();
}
enqueue方法，这里的入队方法比较简单：
- 将尾节点的next指向新增Node
- last节点指向新增Node
private void enqueue(Node<E> node) {
last = last.next = node;
}
signalNotEmpty方法，也很简单，就唤醒消费者去消费。
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
notEmpty.signal();
} finally {
takeLock.unlock();
}
}
offer方法
众所周知，offer有2种方法，一种是非阻塞直接给添加元素结果的，另一种是阻塞一段设定的后返回添加元素的结果。
我们先看看简单的offer方法，非阻塞直接返回和put方法很像，返回了这次添加操作的结果：
public boolean offer(E e) {
if (e == null) throw new NullPointerException();
final AtomicInteger count = this.count;
// 如果队列满了，添加失败，返回fals
if (count.get() == capacity)
return false;
// 预制为-1，失败标识
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();
// 返回是否成功失败，c = -1是预置的失败结果
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);
// 预置数量-1，失败标识
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);
}
// 封装Node入队
enqueue(new Node<E>(e));
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
// 判断是否唤醒，c是添加前的数量，消费者线程
if (c == 0)
signalNotEmpty();
return true;
}
消费者方法源码
take方法
take方法会一直阻塞线程，直到获取到元素，或者被中断：
public E take() throws InterruptedException {
E x;
// 预设失败数量-1
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();
}
// 如果消费之前当前队列满了（c是消费前的Node数量），就唤醒生产者线程
if (c == capacity)
signalNotFull();
return x;
}
dequeue方法，获取队列中的头元素：
private E dequeue() {
Node<E> h = head;
Node<E> first = h.next;
h.next = h; // 指向自己
head = first;
E x = first.item;
first.item = null;
return x;
}
signalNotFull方法，唤醒生产者线程：
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
notFull.signal();
} finally {
putLock.unlock();
}
}
poll方法
我们都知道的poll方法，有2种，一种是非阻塞获取队列头元素，而另一种是阻塞一段时间获取。
下面是普通的非阻塞获取poll()方法：
public E poll() {
final AtomicInteger count = this.count;
// 如果队列为空，直接返回null
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();
// 消费头Node前数量 > 1，说明还要其他的Node可以消费，唤醒其他的被阻塞的消费线程
if (c > 1)
notEmpty.signal();
}
} finally {
takeLock.unlock();
}
// 如果消费之前当前队列满了（c是消费前的Node数量），就唤醒生产者线程
if (c == capacity)
signalNotFull();
return x;
}
当队列为空的话，阻塞一段时间的poll方法，超时返回null。
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;
}
后话
这篇，主播围绕着阻塞队列LinkedBlockingQueue的源码的角度，来看看如何阻塞的对队列进行操作。
比如阻塞、非阻塞、超时阻塞。
聪明的你看完之后，有没有发现LinkedBlockingQueue和之前说的ArrayBlockingQueue的区别呢？
