JUC-LinkedBlockingQueue学习

2019年3月10日 245次阅读 来源: JUC

一.概述

LinkedBlockingQueue是单向链表实现的可选的有界的阻塞队列
1. 队列元素是先进先出FIFO (first-in-first-out)
2. 队列的头元素head是在队列中时间最长的元素，因为最先入队列的，获取操作(poll、peek、take)返回头元素head
3. 队列的尾元素last是在队列中时间最短的元素,因为最后入队列的，新元素插入(offer、put)队列的尾部
4. 可选容量范围capacity的构造方法是防止队列过度扩展的一种方法，如果没有指定最大容量capacity，那么最默认的最大容量为Integer.MAX_VALUE
5. 链接队列的节点都是动态创建的，出队列的节点可以被GC所回收，因此其具有灵活的伸缩性
6. LinkedBlockingQueue可以同时进行入队和出队列，它的入队列和出队列使用的是两个不同的lock对象,因此无论是在入队列还是出队列，都会涉及对元素数量count的并发修改，因此这里使用了一个原子操作类来解决对同一个变量进行并发修改的线程安全问题
《JUC-LinkedBlockingQueue学习》

二.LinkedBlockingQueue对象结构

《JUC-LinkedBlockingQueue学习》

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

《JUC-LinkedBlockingQueue学习》

三.节点元素

所有的元素都通过Node这个静态内部类来进行存储，这与LinkedList的处理方式完全一样
元素节点Node对象的next是下面3种之一：
6. 指向下一个元素
7. 指向当前元素, 即是头元素head的下一个元素head.next
8. null, 意味着是最后的元素

static class Node<E> {
    E item;

    Node<E> next;

    Node(E x) { 
        item = x; 
    }
}

《JUC-LinkedBlockingQueue学习》

四.创建对象实例

  1. 可以在创建时手动指定最大容量capacity，如果没有指定最大容量，那么最默认的最大容量为Integer.MAX_VALUE.
  2. 变量capacity被final修饰，表示初始化后不可变
  3. 队列头节点head有一个不变性：头节点item始终为null -head.item=null
  4. 队列尾节点last也有一个不变性：尾节点next始终为null -last.next=null
// 阻塞队列所能存储的最大容量
private final int capacity;

/** * 链表队列的头节点 * 链表的头部元素item始终为null：head.item = null * 创建实例时：last = head = new Node<E>(null); */
transient Node<E> head;

/** * 链表队列的尾节点 * 链表尾元素下一个元素next始终为null：tail.next = null * 创建实例时：last = head = new Node<E>(null); */
private transient Node<E> last;

/** * 创建一个容量为 Integer.MAX_VALUE 的 LinkedBlockingQueue */
public LinkedBlockingQueue() {
    this(Integer.MAX_VALUE);
}

/** * 创建一个具有给定（固定）容量的 LinkedBlockingQueue * * @param capacity 队列容量大小 * @throws IllegalArgumentException 如果队列容量小于0抛出异常 */
public LinkedBlockingQueue(int capacity) {
    if (capacity <= 0) throw new IllegalArgumentException();
    this.capacity = capacity;
    last = head = new Node<E>(null);
}

/** * 创建一个容量是 Integer.MAX_VALUE 的 LinkedBlockingQueue * 最初包含给定 collection 的元素，元素按该 collection 迭代器的遍历顺序添加。 * * @param c 包含初始元素所属的 collection * @throws NullPointerException 如果指定 collection 或其所有元素均为 null抛出异常 */
public LinkedBlockingQueue(Collection<? extends E> c) {
    this(Integer.MAX_VALUE);
    final ReentrantLock putLock = this.putLock;
    // 获取锁
    putLock.lock(); // Never contended, but necessary for visibility
    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();
    }
}

创建队列实例

// 最大容量10
LinkedBlockingQueue<Integer> linkedQueue1 = new LinkedBlockingQueue<>(10);

// 最大容量Integer.MAX_VALUE
LinkedBlockingQueue<Integer> linkedQueue2 = new LinkedBlockingQueue<>();

// 最大容量Integer.MAX_VALUE，用集合list初始化
List<Integer> list = new ArrayList<>();
LinkedBlockingQueue linkedQueue3 = new LinkedBlockingQueue(list);

五.入队列

  1. 在入队列和出队列时使用的不是同一个Lock，入队列使用ReentrantLock putLock
  2. 入队列使用Condition notFull对象监视器-当队列的元素已经达到最大容量capactiy，通过该notFull让入队列的线程处于等待状态
  3. 当前阻塞队列中的元素数量AtomicInteger count，入队列和出队列使用的是两个不同的lock对象,但无论是在入队列还是出队列，都会涉及对元素数量count的并发修改，因此这里使用了一个原子操作类来解决对同一个变量进行并发修改的线程安全问题
  4. 入队列操作之前需要先获取putLock锁，执行完释放锁putLock.unlock();
  5. 当队列达到最大容量count.get() = capacity，notFull对象监视器notFull.await()等待，即生产线程停止生产，等待队列未满

5.1 定义变量

/** * 当前阻塞队列中的元素数量 * PS:如果看过ArrayBlockingQueue的源码,会发现ArrayBlockingQueue底层保存元素数量使用的是一个普通的int类型变量。 * 其原因是在ArrayBlockingQueue底层对于元素的入队列和出队列使用的是同一个lock对象。而数量的修改都是在处于线程获取锁的情况下进行操作， 因此不会有线程安全问题。 * 而LinkedBlockingQueue却不是，它的入队列和出队列使用的是两个不同的lock对象,因此无论是在入队列还是出队列，都会涉及对元素数量count的并发修改，因此这里使用了一个原子操作类来解决对同一个变量进行并发修改的线程安全问题。*/
private final AtomicInteger count = new AtomicInteger();

/** * 1. 元素入队列时线程所获取的锁 * 2. 当执行add、put、offer等操作时线程需要获取锁 */
private final ReentrantLock putLock = new ReentrantLock();

 /** * 当队列的元素已经达到capactiy，通过该Condition让元素入队列的线程处于等待状态 */
private final Condition notFull = putLock.newCondition();

5.2 入队列 – enqueue

/** * 在队列的尾部插入node * @param node the node */
private void enqueue(Node<E> node) {
    last = last.next = node;
}

上面的代码其实没什么花样，赋值操作符“=”是从右往左的，所以上面代码等价于：

last.next = node
last=node

5.2.1 空队列

创建队列或出队列为空时，头结点head和尾节点last都指向同一节点

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

《JUC-LinkedBlockingQueue学习》

5.2.2 入队列1

last = last.next = node1

《JUC-LinkedBlockingQueue学习》

5.2.3 入队列2

last = last.next = node2

《JUC-LinkedBlockingQueue学习》

5.3 入队列方法

5.3.1 add(E e)

  1. 如果队列未满，立即执行在队列的尾部插入指定的元素e，成功返回true
  2. 如果队列已满，抛出异常IllegalStateException
/** * add(e)测试： * @throws NullPointerException 待插入元素为null * @throws IllegalStateException 如果队列已满时执行插入 */
 public boolean add(E e) {
    if (offer(e))
        return true;
    else
        throw new IllegalStateException("Queue full");
}

5.3.2 offer(E e)

  1. offer方法是非阻塞的， 当队列已经满了，它不会继续等待,而是直接返回
  2. 当入队列获取到锁时，需要进行二次的检查,因为可能当队列的大小为capacity-1时，两个线程同时去抢占锁，而只有一个线程抢占成功，那么此时当线程将元素入队列后，释放锁，后面的线程抢占锁之后，此时队列大小已经达到capacity，所以将它无法让元素入队列
/** * offer(e)测试： * 1. 如果队列未满，立即执行在队列的尾部插入指定的元素e，成功返回true * 2. 如果队列已满，返回false * * offer(e)通常要优于add(e),因为add(e)在队列满了时会抛出IllegalStateException异常 * * @throws NullPointerException 待插入元素为null */
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;
}

5.3.3 offer(E e, long timeout, TimeUnit unit)

  1. 一个限时等待插入操作，即在等待一定的时间内，如果队列有空间可以插入，那么就将元素入队列，然后返回true,如果在过完指定的时间后依旧没有空间可以插入，那么就返回false
  2. 通过timeout和TimeUnit来指定等待的时长，timeout为时间的长度,TimeUnit为时间的单位
/** * offer(E e, long timeout, TimeUnit unit)测试： * 1. 如果队列未满，立即执行在队列的尾部插入指定的元素e，成功返回true * 2. 如果队列已满，等待指定的时间timeout(单位为unit)以使空间变为可用。 * * @throws NullPointerException 待插入元素为null * @throws InterruptedException 在等待时被中断 * */
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.3.4 入队列 – put

put为阻塞方法，直到队列有空余时，才能为队列加入新元素

/** * put(e)测试： * 1. 如果队列未满，立即执行在队列的尾部插入指定的元素e * 2. 如果队列已满，等待队列空间可用后唤醒再插入 * * @throws NullPointerException 待插入元素为null * @throws InterruptedException 如果在等待时被中断 */
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();
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        signalNotEmpty();
}

六.出队列

  1. 在入队列和出队列时使用的不是同一个Lock，出队列使用ReentrantLock takeLock
  2. 出队列使用Condition notEmpty对象监视器-当队列的为空时，通过该notEmpty让出队列的线程处于等待状态
  3. 当前阻塞队列中的元素数量AtomicInteger count，入队列和出队列使用的是两个不同的lock对象,但无论是在入队列还是出队列，都会涉及对元素数量count的并发修改，因此这里使用了一个原子操作类来解决对同一个变量进行并发修改的线程安全问题
  4. 出队列操作之前需要先获取takeLock锁，执行完释放锁takeLock.unlock();
  5. 当队列为空count.get() = 0，notEmpty对象监视器notEmpty.await()等待，即消费线程停止消费，等待队列不为空

6.1 定义变量

private final AtomicInteger count = new AtomicInteger();

/** * 元素出队列时线程所获取的锁 * 当执行take、poll等操作时线程需要获取的锁 */
private final ReentrantLock takeLock = new ReentrantLock();

/** * 当队列为空时，通过该Condition让从队列中获取元素的线程处于等待状态 */
private final Condition notEmpty = takeLock.newCondition();

6.2 出队列流程

/** * 从队列的头部删除节点node * 头部元素出队列的过程,其最终的目的是让原来的head被GC回收，让其next成为head,并且新的head的item为null. * 因为LinkedBlockingQueue的头部具有一致性:即元素为null。 * * @return the node */
private E dequeue() {
    Node<E> h = head;
    Node<E> first = h.next;
    h.next = h; // help GC
    head = first;
    E x = first.item;
    first.item = null;
    return x;
}

6.2.1 满队列

《JUC-LinkedBlockingQueue学习》

6.2.2 出队列1

《JUC-LinkedBlockingQueue学习》

6.2.3 出队列2

《JUC-LinkedBlockingQueue学习》

6.2.4 出队列节点GC

/** * 从队列的头部删除节点node * 头部元素出队列的过程,其最终的目的是让原来的head被GC回收，让其next成为head,并且新的head的item为null. * 因为LinkedBlockingQueue的头部具有一致性:即元素为null */
h.next = h;

《JUC-LinkedBlockingQueue学习》

6.3 出队列方法

6.3.1 poll()

  1. 获取并移除此队列的头，如果此队列为空，则返回 null
  2. 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();
            if (c > 1)
                notEmpty.signal();
        }
    } finally {
        takeLock.unlock();
    }
    if (c == capacity)
        signalNotFull();
    return x;
}

6.3.2 poll(timeout, unit)

  1. 获取并移除此队列的头部，在指定的等待时间前等待可用的元素
  2. pool()为非阻塞方法
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;
}

6.3.3 peek()

  1. 获取但不移除此队列的头；如果此队列为空，则返回 null
  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;
        else
            return first.item;
    } finally {
        takeLock.unlock();
    }
}

6.3.4 take()

  1. 获取并移除此队列的头部，在元素变得可用之前一直等待（如果有必要）
  2. 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;
}

七.迭代器

LinkedBlockingQueue的迭代器，是多线程安全的，在获取元素之前，会对读锁和写锁同时加锁，同时，为了防止死锁，读锁和写锁的加解锁顺序，也是经过设计的：

void fullyLock() {
    putLock.lock();// 先加写锁
    takeLock.lock();// 再加读锁
}

void fullyUnlock() {
    takeLock.unlock();
    putLock.unlock();
}

LinkedBlockingQueue的迭代器中，保存了以下内容：

private Node<E> current; // 指向下一个节点
private Node<E> lastRet; // 记录当前节点
private E currentElement; // 当前节点内容

首先，保存了当前需要返回的内容，可以保证在当前节点移除的情况下，迭代器的next()方法，也能返回当前指向的内容，即使先调用hasNext()方法，其他线程删除了当前对象，那么next()方法也可以保证返回正确对象
其次，如果在迭代器中，调用remove()方法，删除了当前对象，那么 lastRet方法就用上了，可以通过再次遍历列表，找到需要删除的对象，并将其删除，同时为了防止remove()方法被调用两次，在删除时，会将 lastRet设置为null，如果只有这一个指针，那么remove()之后，这个迭代器就啥也干不了了
最后，current保存了迭代器的下一个指向的位置，调用hasNext()时，可以立即直到是否还有空余对象，更重要的是，如果在迭代器创建后，其他线程多次调用了出队的方法，可能导致 lastRet和current都变成悬挂的指针了，这时，只要判断current的next是否为自己，就可以知道自己是否已经被出队，是否需要重定向current的位置

八.LinkedBlockingQueue与ArrayBlockingQueue的比较

  1. ArrayBlockingQueue由于其底层基于数组，并且在创建时指定存储的大小，在完成后就会立即在内存分配固定大小容量的数组元素，因此其存储通常有限，故其是一个“有界“的阻塞队列
  2. LinkedBlockingQueue可以由用户指定最大存储容量，也可以无需指定，如果不指定则最大存储容量将是Integer.MAX_VALUE，即可以看作是一个“无界”的阻塞队列，由于其节点的创建都是动态创建，并且在节点出队列后可以被GC所回收，因此其具有灵活的伸缩性
  3. 由于ArrayBlockingQueue的有界性，因此其能够更好的对于性能进行预测，LinkedBlockingQueue由于没有限制大小，当任务非常多的时候，不停地向队列中存储，就有可能导致内存溢出的情况发生
  4. ArrayBlockingQueue中在入队列和出队列操作过程中，使用的是同一个lock，所以即使在多核CPU的情况下，其读取和操作的都无法做到并行
  5. LinkedBlockingQueue的读取和插入操作所使用的锁是两个不同的lock，它们之间的操作互相不受干扰，因此两种操作可以并行完成，故LinkedBlockingQueue的吞吐量要高于ArrayBlockingQueue

九.总结

JDK中选用LinkedBlockingQueue作为阻塞队列的原因：
就在于其无界性。因为线程大小固定的线程池，其线程的数量是不具备伸缩性的，当任务非常繁忙的时候，就势必会导致所有的线程都处于工作状态
1. 使用一个有界的阻塞队列来进行处理，那么就非常有可能很快导致队列满的情况发生，从而导致任务无法提交而抛出RejectedExecutionException
2. 使用一个无界队列由于其良好的存储容量的伸缩性，可以很好的去缓冲任务繁忙情况下场景，即使任务非常多，也可以进行动态扩容，当任务被处理完成之后，队列中的节点也会被随之被GC回收，非常灵活

十.参考

LinkedBlockingQueue源码分析
https://zhidao.baidu.com/question/393766722974951645.html

    原文作者：JUC
    原文地址: https://blog.csdn.net/lipinganq/article/details/78629427
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞