ArrayBlockingQueue源码分析

思想

阻塞队列就是典型的生产者-消费者模型，它可以做到以下几点:

1. 当阻塞队列数据为空时，所有的消费者线程都会被阻塞，等待队列非空。
2. 当生产者往队列里填充数据后，队列就会通知消费者队列非空，消费者此时就可以进来消费。
3. 当阻塞队列因为消费者消费过慢或者生产者存放元素过快导致队列填满时无法容纳新元素时，生产者就会被阻塞，等待队列非满时继续存放元素。
4. 当消费者从队列中消费一个元素之后，队列就会通知生产者队列非满，生产者可以继续填充数据了。

常见应用在线程池中

最常见的例子无非是我们的线程池,从源码中我们就能看出当核心线程无法及时处理任务时，这些任务都会扔到 workQueue 中。

public ThreadPoolExecutor(int corePoolSize,
                            int maximumPoolSize,
                            long keepAliveTime,
                            TimeUnit unit,
                            BlockingQueue<Runnable> workQueue,
                            ThreadFactory threadFactory,
                            RejectedExecutionHandler handler) {// ...}
1
2
3
4
5
6
7

例子

我们这里会用两个线程分别模拟生产者和消费者，生产者生产完会使用 put 方法生产 10 个元素给消费者进行消费，当队列元素达到我们设置的上限 5 时，put 方法就会阻塞。 同理消费者也会通过 take 方法消费元素，当队列为空时，take 方法就会阻塞消费者线程。这里笔者为了保证消费者能够在消费完 10 个元素后及时退出。便通过倒计时门闩，来控制消费者结束，生产者在这里只会生产 10 个元素。当消费者将 10 个元素消费完成之后，按下倒计时门闩，所有线程都会停止。

public class ProducerConsumerExample {

    public static void main(String[] args) throws InterruptedException {

        // 创建一个大小为 5 的 ArrayBlockingQueue
        ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);

        // 创建生产者线程
        Thread producer = new Thread(() -> {
            try {
                for (int i = 1; i <= 10; i++) {
                    // 向队列中添加元素，如果队列已满则阻塞等待
                    queue.put(i);
                    System.out.println("生产者添加元素：" + i);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        });

        CountDownLatch countDownLatch = new CountDownLatch(1);

        // 创建消费者线程
        Thread consumer = new Thread(() -> {
            try {
                int count = 0;
                while (true) {

                    // 从队列中取出元素，如果队列为空则阻塞等待
                    int element = queue.take();
                    System.out.println("消费者取出元素：" + element);
                    ++count;
                    if (count == 10) {
                        break;
                    }
                }

                countDownLatch.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        });

        // 启动线程
        producer.start();
        consumer.start();

        // 等待线程结束
        producer.join();
        consumer.join();

        countDownLatch.await();

        producer.interrupt();
        consumer.interrupt();
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

结构

add

//ArrayBlockingQueue 
public boolean add(E e) {
    return super.add(e);
}

//BlockingQueue 
public boolean add(E e) {
    //AbstractQueue的offer来自下层的ArrayBlockingQueue从BlockingQueue继承并实现的offer方法
    if (offer(e))
        return true;
    else
        throw new IllegalStateException("Queue full");
}

//ArrayBlockingQueue 
public boolean offer(E e) {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        if (count == items.length)
            return false;
        else {
            enqueue(e);
            return true;
        }
    } finally {
        lock.unlock();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

构造方法

// capacity 表示队列初始容量，fair 表示 锁的公平性
public ArrayBlockingQueue(int capacity, boolean fair) {
									//如果设置的队列大小小于0，则直接抛出IllegalArgumentException
        if (capacity <= 0)
            throw new IllegalArgumentException();
        //初始化一个数组用于存放队列的元素
        this.items = new Object[capacity];
        //创建阻塞队列流程控制的锁
        lock = new ReentrantLock(fair);
        //用lock锁创建两个条件控制队列生产和消费
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14

可以看到两个重入锁 非空和非满，它们是实现生产者和消费者有序工作的关键所在。ArrayBlockingQueue 用的是非公平锁，即各个生产者或者消费者线程收到通知后，对于锁的争抢是随机的。

增删查

阻塞式获取和新增元素

ArrayBlockingQueue 阻塞式获取和新增元素的方法为：

• put(E e)：将元素插入队列中，如果队列已满，则该方法会一直阻塞，直到队列有空间可用或者线程被中断。
• take() ：获取并移除队列头部的元素，如果队列为空，则该方法会一直阻塞，直到队列非空或者线程被中断。

非空条件对象

假设我们的代码消费者先启动，当它发现队列中没有数据，那么非空条件就会将这个线程挂起，即等待条件非空时挂起。然后 CPU 执行权到达生产者，生产者发现队列中可以存放数据，于是将数据存放进去，通知此时条件非空，此时消费者就会被唤醒到队列中使用 take 等方法获取值了。

非满条件对象

生产者将队列塞满后再次尝试将数据存入队列，发现队列已满，于是阻塞队列就将当前线程挂起，等待非满。然后消费者拿着 CPU 执行权进行消费，于是队列可以存放新数据了，发出一个非满的通知，此时挂起的生产者就会等待 CPU 执行权到来时再次尝试将数据存到队列中。

put 源码

public void put(E e) throws InterruptedException {
    //确保插入的元素不为null
    checkNotNull(e);
    //加锁
    final ReentrantLock lock = this.lock;
    //这里使用lockInterruptibly()方法而不是lock()方法是为了能够响应中断操作，如果在等待获取锁的过程中被打断则该方法会抛出InterruptedException异常。
    lock.lockInterruptibly();
    try {
   				 //如果count等数组长度则说明队列已满，当前线程将被挂起放到AQS队列中，等待队列非满时插入（非满条件）。
       //在等待期间，锁会被释放，其他线程可以继续对队列进行操作。
        while (count == items.length)
            notFull.await();
    			 //如果队列可以存放元素，则调用enqueue将元素入队
        enqueue(e);
    } finally {
        //释放锁
        lock.unlock();
    }
}
private void enqueue(E x) {
   //获取队列底层的数组
    final Object[] items = this.items;
    //将putindex位置的值设置为我们传入的x
    items[putIndex] = x;
    //更新putindex，如果putindex等于数组长度，则更新为0
    if (++putIndex == items.length)
        putIndex = 0;
    //队列长度+1
    count++;
    //通知队列非空，那些因为获取元素而阻塞的线程可以继续工作了
    notEmpty.signal();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

notEmpty.signal() 通知队列非空，消费者可以从队列中获取值了。

take 源码

public E take() throws InterruptedException {
    //获取锁
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
    				//如果队列中元素个数为0，则将当前线程打断并存入AQS队列中，等待队列非空时获取并移除元素（非空条件）
        while (count == 0)
            notEmpty.await();
    			 //如果队列不为空则调用dequeue获取元素
        return dequeue();
    } finally {
      	 //释放锁
        lock.unlock();
    }
}
private E dequeue() {
  	//获取阻塞队列底层的数组
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    //从队列中获取takeIndex位置的元素
    E x = (E) items[takeIndex];
    //将takeIndex置空
    items[takeIndex] = null;
    //takeIndex向后挪动，如果等于数组长度则更新为0
    if (++takeIndex == items.length)
        takeIndex = 0;
    //队列长度减1
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    //通知那些被打断的线程当前队列状态非满，可以继续存放元素
    notFull.signal();
    return x;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

非空非满交互

非阻塞式获取和新增元素

offer(E e)：将元素插入队列尾部。如果队列已满，则该方法会直接返回 false，不会等待并阻塞线程。
poll()：获取并移除队列头部的元素，如果队列为空，则该方法会直接返回 null，不会等待并阻塞线程。
add(E e)：将元素插入队列尾部。如果队列已满则会抛出 IllegalStateException 异常，底层基于 offer(E e) 方法。
remove()：移除队列头部的元素，如果队列为空则会抛出 NoSuchElementException 异常，底层基于 poll()。
peek()：获取但不移除队列头部的元素，如果队列为空，则该方法会直接返回 null，不会等待并阻塞线程。

指定超时时间内阻塞式获取和新增元素

在 offer(E e) 和 poll() 非阻塞获取和新增元素的基础上，设计者提供了带有等待时间的 offer(E e, long timeout, TimeUnit unit) 和 poll(long timeout, TimeUnit unit) ，用于在指定的超时时间内阻塞式地添加和获取元素。

public boolean offer(E e, long timeout, TimeUnit unit)
 throws InterruptedException {

  checkNotNull(e);
  long nanos = unit.toNanos(timeout);
  final ReentrantLock lock = this.lock;
  lock.lockInterruptibly();
  try {
  //队列已满，进入循环
      while (count == items.length) {
      //时间到了队列还是满的，则直接返回false
          if (nanos <= 0)
              return false;
           //阻塞nanos时间，等待非满
          nanos = notFull.awaitNanos(nanos);
      }
      enqueue(e);
      return true;
  } finally {
      lock.unlock();
  }
}


public E poll(long timeout, TimeUnit unit) throws InterruptedException {
   long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
    	//队列为空，循环等待，若时间到还是空的，则直接返回null
        while (count == 0) {
            if (nanos <= 0)
                return null;
            nanos = notEmpty.awaitNanos(nanos);
        }
        return dequeue();
    } finally {
        lock.unlock();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

可以看到，带有超时时间的 offer 方法在队列已满的情况下，会等待用户所传的时间段，如果规定时间内还不能存放元素则直接返回 false。带有超时时间的 poll 也一样，队列为空则在规定时间内等待，若时间到了还是空的，则直接返回 null。

判断元素是否存在

ArrayBlockingQueue 提供了 contains(Object o) 来判断指定元素是否存在于队列中。

public boolean contains(Object o) {
    //若目标元素为空，则直接返回 false
    if (o == null) return false;
    //获取当前队列的元素数组
    final Object[] items = this.items;
    //加锁
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 如果队列非空
        if (count > 0) {
            final int putIndex = this.putIndex;
            //从队列头部开始遍历
            int i = takeIndex;
            do {
                if (o.equals(items[i]))
                    return true;
                if (++i == items.length)
                    i = 0;
            } while (i != putIndex);
        }
        return false;
    } finally {
        //释放锁
        lock.unlock();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

面试题

ArrayBlockingQueue 是什么？它的特点是什么？

ArrayBlockingQueue 是 BlockingQueue 接口的有界队列实现类，常用于多线程之间的数据共享，底层采用数组实现，从其名字就能看出来了。
ArrayBlockingQueue 的容量有限，一旦创建，容量不能改变。
为了保证线程安全，ArrayBlockingQueue 的并发控制采用可重入锁 ReentrantLock ，不管是插入操作还是读取操作，都需要获取到锁才能进行操作。并且，它还支持公平和非公平两种方式的锁访问机制，默认是非公平锁。
ArrayBlockingQueue 虽名为阻塞队列，但也支持非阻塞获取和新增元素（例如 poll() 和 offer(E e) 方法），只是队列满时添加元素会抛出异常，队列为空时获取的元素为 null，一般不会使用。

ArrayBlockingQueue 和 LinkedBlockingQueue 有什么区别？

ArrayBlockingQueue 和 LinkedBlockingQueue 是 Java 并发包中常用的两种阻塞队列实现，它们都是线程安全的。
不过，不过它们之间也存在下面这些区别：底层实现：ArrayBlockingQueue 基于数组实现，而 LinkedBlockingQueue 基于链表实现。
是否有界：ArrayBlockingQueue 是有界队列，必须在创建时指定容量大小。LinkedBlockingQueue 创建时可以不指定容量大小，默认是Integer.MAX_VALUE，也就是无界的。但也可以指定队列大小，从而成为有界的。
锁是否分离： ArrayBlockingQueue中的锁是没有分离的，即生产和消费用的是同一个锁；LinkedBlockingQueue中的锁是分离的，即生产用的是putLock，消费是takeLock，这样可以防止生产者和消费者线程之间的锁争夺。
内存占用：ArrayBlockingQueue 需要提前分配数组内存，而 LinkedBlockingQueue 则是动态分配链表节点内存。这意味着，ArrayBlockingQueue 在创建时就会占用一定的内存空间，且往往申请的内存比实际所用的内存更大，而LinkedBlockingQueue 则是根据元素的增加而逐渐占用内存空间。

ArrayBlockingQueue 和 ConcurrentLinkedQueue 有什么区别？

ArrayBlockingQueue 和 ConcurrentLinkedQueue 是 Java 并发包中常用的两种队列实现，它们都是线程安全的。不过，不过它们之间也存在下面这些区别：
底层实现：ArrayBlockingQueue 基于数组实现，而 ConcurrentLinkedQueue 基于链表实现。
是否有界：ArrayBlockingQueue 是有界队列，必须在创建时指定容量大小，而 ConcurrentLinkedQueue 是无界队列，可以动态地增加容量。
是否阻塞：ArrayBlockingQueue 支持阻塞和非阻塞两种获取和新增元素的方式（一般只会使用前者）， ConcurrentLinkedQueue 是无界的，仅支持非阻塞式获取和新增元素。

ArrayBlockingQueue 的实现原理是什么？

ArrayBlockingQueue 的实现原理主要分为以下几点（这里以阻塞式获取和新增元素为例介绍）：
ArrayBlockingQueue 内部维护一个定长的数组用于存储元素。
通过使用 ReentrantLock 锁对象对读写操作进行同步，即通过锁机制来实现线程安全。
通过 Condition 实现线程间的等待和唤醒操作。
这里再详细介绍一下线程间的等待和唤醒具体的实现（不需要记具体的方法，面试中回答要点即可）：
当队列已满时，生产者线程会调用 notFull.await() 方法让生产者进行等待，等待队列非满时插入（非满条件）。
当队列为空时，消费者线程会调用 notEmpty.await()方法让消费者进行等待，等待队列非空时消费（非空条件）。
当有新的元素被添加时，生产者线程会调用 notEmpty.signal()方法唤醒正在等待消费的消费者线程。
当队列中有元素被取出时，消费者线程会调用 notFull.signal()方法唤醒正在等待插入元素的生产者线程。

补充Condition（对象监视器）

比如可以实现多路通知功能也就是在一个Lock对象中可以创建多个Condition实例（即对象监视器），线程对象可以注册在指定的Condition中，从而可以有选择性的进行线程通知，在调度线程上更加灵活。
在使用notify()/notifyAll()方法进行通知时，被通知的线程是由 JVM 选择的，用ReentrantLock类结合Condition实例可以实现“选择性通知” ，这个功能非常重要，而且是 Condition 接口默认提供的。
而synchronized关键字就相当于整个 Lock 对象中只有一个Condition实例，所有的线程都注册在它一个身上。如果执行notifyAll()方法的话就会通知所有处于等待状态的线程，这样会造成很大的效率问题。
而Condition实例的signalAll()方法，只会唤醒注册在该Condition实例中的所有等待线程。

作者声明

如有问题，欢迎指正！
1