并发和多线程(二十)–LinkedBlockingQueue源码解析

目录

• 类定义
  
  
  
  • Queue
  
  
  • BlockingQueue
  
  
  
  


• 成员变量


• 构造函数


• offer()


• put()


• take()


• enqueue and dequeue


• peek()

阻塞队列在日常开发中直接使用比较少，但是在很多工具类或者框架中有很多应用，例如线程池，消息队列等。所以，深入了解阻塞队列也是很有必要的。所以这里来了解一下LinkedBlockingQueue的相关源码，从命名可以看出来是由链表实现的数据结构。

类定义

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


从上面可以看到继承Queue，实现BlockingQueue，我们介绍一下两个类里面方法的作用，为了方便记忆和对比放到表格里进行展示。

Queue

BlockingQueue

BlockingQueue顾名思义带有阻塞的队列，方法有所区别，下面方法包含了Queue，因为属于继承关系，下面表格方法名用序号代替。

成员变量

//链表的容量，默认Integer.MAX_VALUE
private final int capacity;
//当前存在元素数量
private final AtomicInteger count = new AtomicInteger();
//链表的head节点
transient Node head;
//链表的tail节点
private transient Node 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比较简单，一个item，还有指向下个节点，也就是单向链表
static class Node {
E item;
/**
* One of:
* - the real successor Node
* - this Node, meaning the successor is head.next
* - null, meaning there is no successor (this is the last node)
*/
Node next;
Node(E x) { item = x; }
}


构造函数

//默认队列存储的元素最大为Integer.MAX_VALUE
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}

//自定义capacity，初始化head、tail
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
last = head = new Node(null);
}

public LinkedBlockingQueue(Collection c) {
this(Integer.MAX_VALUE);
final ReentrantLock putLock = this.putLock;
//加锁，因为是添加，肯定是putLock
putLock.lock();
try {
int n = 0;
//遍历集合，每次生成一个node，添加到链表尾部
for (E e : c) {
if (e == null)
throw new NullPointerException();
//每次判断新增的节点是否超过capacity，如果是，抛出异常
if (n == capacity)
throw new IllegalStateException("Queue full");
//将节点添加到队列tail
enqueue(new Node(e));
++n;
}
//设置当前元素个数count
count.set(n);
//finally解锁
} finally {
putLock.unlock();
}
}


offer()

public boolean offer(E e) {
if (e == null) throw new NullPointerException();
final AtomicInteger count = this.count;
if (count.get() == capacity)
return false;
//初始设置为-1，c <0,表示新增失败
int c = -1;
Node node = new Node(e);
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
if (count.get() enqueue(node);
c = count.getAndIncrement();
if (c + 1 notFull.signal();
}
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
return c >= 0;
}


流程：

1. 如果e为空，抛出异常。


2. 如果当前队列已满，返回false。


3. 将e封装成一个新的节点，加锁，putLock。


4. 再次判断队列元素数量
   
5. CAS将count+1，注意这里调用的是getAndIncrement返回的是+1之前的值。如果队列没满，唤醒某个某个因为添加而阻塞的线程。


6. finally解锁，如果c == 0，加锁takeLock，唤醒继续添加。


7. 返回 c >= 0。

put()

相对于offer()，put的代码会判断当前队列是否满了，如果满了，通过Condition阻塞，其他没啥区别。

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();
}
//注意这里c是先获取，后-1的
if (c == capacity)
signalNotFull();
return x;
}


流程：

1. 加锁，takeLock。


2. 如果当前队列为空，直接通过notEmpty阻塞，等待被唤醒。


3. 取出第一个元素，并删除元素。


4. 如果c > 1，表示队列还有元素，唤醒别的线程获取。


5. finally解锁，如果c == capacity，表示队列没满，加锁takeLock，唤醒继续添加。


6. 返回 x。

enqueue and dequeue

private void enqueue(Node node) {
last = last.next = node;
}
private E dequeue() {
Node h = head;
Node first = h.next;
h.next = h; // help GC
head = first;
E x = first.item;
first.item = null;
return x;
}


这里统一讲一下在链表中添加和删除数据的流程，特别是dequeue()，我刚看第一眼的时候有点蒙蔽的，下面举个栗子。

BlockingQueue queue = new LinkedBlockingDeque<>();
queue.offer(1);
queue.offer(2);
queue.offer(3);
Integer take = queue.take();
System.out.println(take);


这里把每一步都画出来了，还是比较好理解的。其余方法的逻辑都比较相似，下面简单说一下。

peek()

peek()和take()的代码差不多，只是不会删除元素，take()通过dequeue()，而peek()通过一句代码Node first = head.next;获得该节点的数据然后返回。