LinkedBlockingQueue介绍
【1】LinkedBlockingQueue是一个基于链表实现的阻塞队列，默认情况下，该阻塞队列的大小为Integer.MAX_VALUE，由于这个数值特别大，所以 LinkedBlockingQueue 也被称作无界队列，代表它几乎没有界限，队列可以随着元素的添加而动态增长，但是如果没有剩余内存，则队列将抛出OOM错误。所以为了避免队列过大造成机器负载或者内存爆满的情况出现，我们在使用的时候建议手动传一个队列的大小。
【2】LinkedBlockingQueue内部由单链表实现，只能从head取元素，从tail添加元素。LinkedBlockingQueue采用两把锁的锁分离技术实现入队出队互不阻塞，添加元素和获取元素都有独立的锁，也就是说LinkedBlockingQueue是读写分离的，读写操作可以并行执行。
LinkedBlockingQueue使用
//指定队列的大小创建有界队列 BlockingQueue<Integer> boundedQueue = new LinkedBlockingQueue<>(100); //无界队列 BlockingQueue<Integer> unboundedQueue = new LinkedBlockingQueue<>();
LinkedBlockingQueue的源码分析
【1】属性值
// 容量,指定容量就是有界队列
private final int capacity;
// 元素数量，用原子操作类的原因在于有两个线程都会操作需要保证可见性
private final AtomicInteger count = new AtomicInteger();
// 链表头 本身是不存储任何元素的，初始化时item指向null
transient Node<E> head;
// 链表尾
private transient Node<E> last;
// take锁 锁分离，提高效率
private final ReentrantLock takeLock = new ReentrantLock();
// notEmpty条件
// 当队列无元素时，take锁会阻塞在notEmpty条件上，等待其它线程唤醒
private final Condition notEmpty = takeLock.newCondition();
// put锁
private final ReentrantLock putLock = new ReentrantLock();
// notFull条件
// 当队列满了时，put锁会会阻塞在notFull上，等待其它线程唤醒
private final Condition notFull = putLock.newCondition();
//典型的单链表结构
static class Node<E> {
E item; //存储元素
Node<E> next; //后继节点 单链表结构
Node(E x) { item = x; }
}
【2】构造函数
public LinkedBlockingQueue() {
// 如果没传容量，就使用最大int值初始化其容量
this(Integer.MAX_VALUE);
}
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
// 初始化head和last指针为空值节点
last = head = new Node<E>(null);
}
public LinkedBlockingQueue(Collection<? extends E> c) {
this(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();
}
}
【3】核心方法分析
1）入队put方法
public void put(E e) throws InterruptedException {
// 不允许null元素
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;
// 使用put锁加锁
putLock.lockInterruptibly();
try {
// 如果队列满了，就阻塞在notFull上等待被其它线程唤醒（阻塞生产者线程）
while (count.get() == capacity) {
notFull.await();
}
// 队列不满，就入队
enqueue(node);
c = count.getAndIncrement();// 队列长度加1，返回原值
// 如果现队列长度小于容量，notFull条件队列转同步队列，准备唤醒一个阻塞在notFull条件上的线程(可以继续入队)
// 这里为啥要唤醒一下呢？
// 因为可能有很多线程阻塞在notFull这个条件上,而取元素时只有取之前队列是满的才会唤醒notFull,此处不用等到取元素时才唤醒
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock(); // 真正唤醒生产者线程
}
// 如果原队列长度为0，现在加了一个元素后立即唤醒阻塞在notEmpty上的线程
if (c == 0)
signalNotEmpty();
}
private void enqueue(Node<E> node) {
// 直接加到last后面,last指向入队元素
last = last.next = node;
}
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();// 加take锁
try {
notEmpty.signal();// notEmpty条件队列转同步队列，准备唤醒阻塞在notEmpty上的线程
} finally {
takeLock.unlock(); // 真正唤醒消费者线程
}
}
2）出队take方法
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
// 使用takeLock加锁
takeLock.lockInterruptibly();
try {
// 如果队列无元素，则阻塞在notEmpty条件上（消费者线程阻塞）
while (count.get() == 0) {
notEmpty.await();
}
// 否则，出队
x = dequeue();
c = count.getAndDecrement();//长度-1，返回原值
if (c > 1)// 如果取之前队列长度大于1，notEmpty条件队列转同步队列，准备唤醒阻塞在notEmpty上的线程，原因与入队同理
notEmpty.signal();
} finally {
takeLock.unlock(); // 真正唤醒消费者线程
}
// 为什么队列是满的才唤醒阻塞在notFull上的线程呢？
// 因为唤醒是需要加putLock的，这是为了减少锁的次数,所以，这里索性在放完元素就检测一下，未满就唤醒其它notFull上的线程,
// 这也是锁分离带来的代价
// 如果取之前队列长度等于容量（已满），则唤醒阻塞在notFull的线程
if (c == capacity)
signalNotFull();
return x;
}
private E dequeue() {
// head节点本身是不存储任何元素的
// 这里把head删除，并把head下一个节点作为新的值
// 并把其值置空，返回原来的值
Node<E> h = head;
Node<E> first = h.next;
h.next = h; // 方便GC
head = first;
E x = first.item;
first.item = null;
return x;
}
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
notFull.signal();// notFull条件队列转同步队列，准备唤醒阻塞在notFull上的线程
} finally {
putLock.unlock(); // 解锁，这才会真正的唤醒生产者线程
}
}
LinkedBlockingQueue总结
【1】无界阻塞队列，可以指定容量，默认为 Integer.MAX_VALUE，先进先出，存取互不干扰
【2】数据结构：链表（可以指定容量，默认为 Integer.MAX_VALUE，内部类Node存储元素）
【3】锁分离：存取互不干扰，存取操作的是不同的Node对象（takeLock【取Node节点保证前驱后继不乱】，putLock【存Node节点保证前驱后继不乱】，删除时则两个锁一起加）【这是最大的亮点】
【4】阻塞对象（notEmpty【出队：队列count=0，无元素可取时，阻塞在该对象上】，notFull【入队：队列count=capacity，放不进元素时，阻塞在该对象上】）
【5】入队，从队尾入队，由last指针记录。
【6】出队，从队首出队，由head指针记录。
【7】线程池中采用LinkedBlockingQueue而不采用ArrayBlockingQueue的原因便是因为锁分离带来了性能的提升，大大提高队列的吞吐量。
标签：
留言评论