17、ArrayBlockingQueue和LinkedBlockingQueue源码2分钟看完

1、BlockingQueue

BlockingQueue是JUC包下提供的一个阻塞队列 接口;

1、接口方法

 

队列操作

  • 抛出异常:add(e)、remove()、element()
  • 返回特定值:offer()队尾入队/poll()删除队头元素/peek()
  • 一直阻塞:put(e)/take()
  • 超时退出:offer(e,time,unit)/poll(time,unit)

其中:BlockingQueue 不接受 null 元素。试图 add 、 put 或 offer ⼀个 null 元素时,某些实现会抛出 NullPointerException 。

 

2、阻塞队列分类

  • ArrayBlockingQueue:由数组结构组成的有界阻塞队列。
  • LinkedBlockingQueue:由链表组成的有界阻塞队列。
  • PriorityBlockingQueue:一个支持优先级排序的无界阻塞队列(默认升序排序)。
  • DelayQueue:支持延时获取元素的无界阻塞队列。

队列使用PriorityQueue来实现。

  • SynchronousQueue:一个不存储元素的阻塞队列。
  • LinkedTransferQueue:由链表组成的无界阻塞队列。

其多了tryTransfer()方法和transfer()方法。

  • transfer():可以把生产者传入的元素立刻传输给消费者。如果没有consumer在等待接收元素,transfer方法会将元素存放在队列的tail节点,并等到该元素被消费者消费了才返回。
  • tryTransfer():用来试探生产者传入的元素是否能直接传给消费者。不管是否有consumer正在等待接收元素,都立刻返回。

2、ArrayBlockingQueue

ArrayBlockingQueue是由数组结构组成的有界阻塞队列。通过ReentrantLock保证线程安全、并实现 Producer-Consumer模式。

1、构造函数

 

从构造函数可知,默认采用非公平锁;

public ArrayBlockingQueue(int capacity, boolean fair) {

    if (capacity <= 0)
        throw new IllegalArgumentException();
    // 底层使用对象数组保存元素
    this.items = new Object[capacity];
    // 初始化需要加锁使用的ReentrantLock实例,默认采用非公平锁
    lock = new ReentrantLock(fair);
    /**
     * 判断队列是 空 or 满
     *     notEmpty⽤于执⾏take时进⾏await()等待操作,put时进⾏signal()唤醒操作
     *     notFull⽤于执⾏take时进⾏signal()唤醒操作,put时进⾏await()等待操作
     */
    notEmpty = lock.newCondition();
    notFull =  lock.newCondition();
}

方法释义:

  • 构造函数的入参capacity指定了底层存储元素数组⻓度的⼤⼩;
  • 初始化需要加锁使⽤的ReentrantLock实例,默认采⽤的是⾮公平锁;
  • 基于Lock的Condition判断队列是 空 or 满

    • notEmpty⽤于执⾏take时进⾏await()等待操作,put时进⾏signal()唤醒操作;
    • notFull⽤于执⾏take时进⾏signal()唤醒操作,put时进⾏await()等待操作;

2、put()入队

public void put(E e) throws InterruptedException {

    // 入参不能为空
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    // 加可中断锁
    lock.lockInterruptibly();
    try {

        while (count == items.length)
            // 队列满了,则阻塞等待signal唤醒,同时释放次有的锁。
            notFull.await();
        // 入队操作
        enqueue(e);
    } finally {

        lock.unlock();
    }
}

private void enqueue(E x) {

    final Object[] items = this.items;
    items[putIndex] = x;
    // 如果数据已经插入到数组末尾,则重置putIndex为0,从0开始继续插入。
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
    // 通知take线程解除阻塞
    notEmpty.signal();
}

方法释义:

  • ⾸先尝试获得可中断锁,即:lock.lockInterruptibly(),当执⾏interrupt操作时,该锁可以被中断。
    • 如果数组中元素的个数(count)等于数组的⻓度了,说明队列已经满了,在该线程上执⾏等待操作:notFull.await();,等待signal唤醒。
  • 如果队列没有满,则调⽤enqueue(e)⽅法执⾏⼊列操作;

    1. ⼊列操作⾸先会将待插⼊值x放⼊数组下标为putIndex的位置上,然后再将putIndex加1,来指向下⼀次插⼊的下标位置。

      • 如果加1后的putIndex等于了数组的⻓度,那么说明已经越界了(因为putIndex是从0开始的);则做循环式插⼊,重置putIndex为0,从0开始继续插入。
  • 最后,执⾏count++来计算元素总个数;并调⽤notEmpty.signal()⽅法来解除阻塞;

    • 当队列为空的时候,执⾏take⽅法会被notEmpty.await()阻塞;
    • 此处就是通知take线程解除阻塞。

3、take()出队

public E take() throws InterruptedException {

    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {

        while (count == 0)
            // 如果队列为空,则线程阻塞 等待signal唤醒,释放持有的锁
            notEmpty.await();
        // 执行出队操作
        return dequeue();
    } finally {

        lock.unlock();
    }
}

private E dequeue() {

    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    // 获取takeIndex下标的元素
    E x = (E) items[takeIndex];
    // 将takeIndex下标下的袁术置为null,便于后面GC回收
    items[takeIndex] = null;
    // 如果出队操作的到了数组末尾,则重置takeIndex,从0开始继续取出
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    // 默认itrs为null,不会走进if代码段。
    if (itrs != null)
        itrs.elementDequeued();
    // 通知put线程解除阻塞
    notFull.signal();
    return x;
}

方法释义:

  • take⽅法和put⽅法类似,区别是出队的指针是takeIndex;
  • ⾸先尝试获得可中断锁,即:lock.lockInterruptibly(),当执⾏interrupt操作时,该锁可以被中断。

    • 如果队列中为空;执⾏notEmpty.await()将线程阻塞 等待signal唤醒,释放持有的锁。

    • 当调⽤put⽅法向队列中放⼊元素之后 ,会调⽤notEmpty.signal⽅法对等待的线程执⾏唤醒操作;

  • 如果出队操作的到了数组末尾,则重置takeIndex,从0开始继续取出;

    • 出队执⾏完毕后,调⽤notFull.signal⽅法来唤醒在notFull上⾯

    await的线程。 通知put线程解除阻塞。

3、LinkedBlockingQueue

LinkedBlockingQueue是由链表结构组成的有界阻塞队列。通过ReentrantLock保证线程安全、并实现 Producer-Consumer模式。

1、构造函数

如果不指定容量,则默认LinkedBlockingQueue是无界阻塞队列(capacity = Integer.MAX_VALUE)

&nbsp;

构造函数中会创建⼀个空的节点,作为整个链表的头节点。

2、put()入队

&nbsp;

private void enqueue(Node<E> node) {

    // assert putLock.isHeldByCurrentThread();
    // assert last.next == null;
    last = last.next = node;
}

整个put()流程和ArrayBlockingQueue基本一致,对于链表容量的统计会额外采用一个AtomiceInteger类型的变量count维护。

    • 最后唤醒put()线程的代码段上有一个 c == 0的判断,这里的c是入队操作之前的数量。
/** Current number of elements */
private final AtomicInteger count = new AtomicInteger();

/** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();

/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();

/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();

/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();

这里有个比较有意思的点:

  • 入队操作添加完元素之后,如果发现当前队列的元素数量还没到最大容量,则尝试唤醒其他put()操作阻塞的线程;
if (c + 1 < capacity)
    notFull.signal();

3、take()出队

&nbsp;

private E dequeue() {

    // assert takeLock.isHeldByCurrentThread();
    // assert head.item == null;
    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;
}

整个take()流程和ArrayBlockingQueue基本一致,稍微看一下即可。

  • 最后唤醒take()线程的代码段上有一个 c == capacity的判断,这里的c是出队操作之前的数量。

和LinkedBlockingQueue的put()操作一样:

  • 出队操作移除完元素之后,如果发现当前队列的元素数量 > 1,则尝试唤醒其他take()操作阻塞的线程;
if (c > 1)
    notEmpty.signal();