ArrayBlockingQueue原理详解

个人技术

ArrayBlockingQueue原理详解

介绍

ArrayBlockingQueue是基于数组实现的共享通道，为什么说是共享通道，假说线程A希望给线程B发一个消息，用什么方式来告知线程B是比较合适的呢？可以使用BlockingQueue来实现。

通过上图中的继承关系我们可以清晰的发现ArrayBlockingQueue是BlockingQueue接口的实现，通过名称可以得出它是基于数组实现的，所以更适合做有界的队列。

​ 刚才也提到过它是作为数据共享来进行的线程间数据的传递，那么问题来了，当队列中为空的时候消费队列的线程如何知道队列中有新的元素添加进去，队列满的时候又如何进行处理？我们带着上面的疑问来进行下面的分析。

主要的队列的入队、出队操作如下表所示：

插入队列方法

方法名称 参数描述 返回值 异常信息
add 插入对象 ture代表插入成功，如果队列已满，抛出异常 IllegalStateException(“Queue full”)异常——AbstractQueue
offer 插入对象 true代表插入成功，队列已满直接返回false 无
offer 插入对象，等待时间 true代表插入成功，队列已满等待一段时间后仍没有空间则返回false 无
put 插入对象 true代表插入成功，如果队列已满则阻塞线程等待队列为空的时候插入

获取队列内容

方法名称 参数描述 返回值 异常信息
remove 无 返回队首数据并移除，队列已空则抛出异常信息 NoSuchElementException()异常——AbstractQueue
poll 无 列不为空时返回队首值并移除；队列为空时返回null。非阻塞立即返回。
poll 等待时间 设定等待的时间，如果在指定时间内队列还未孔则返回null，不为空则返回队首值
take 无 队列不为空返回队首值并移除；当队列为空时会阻塞等待，一直等到队列不为空时再返回队首值。

简单实例

上面的方法中重点的内容在于put和take方法，我们以一个实例来看一下这个队列的作用。

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
/**
 * ArrayBlockingQueue内容测试demo
 *
 * @author <a href="mailto:dwlsxj@126.com">battleheart</a>
 */
public class BlockQueueDemo {
    public static final ArrayBlockingQueue<Integer> arr = new ArrayBlockingQueue<>(10);

    private static int sum = 0;

    public static void main(String[] args) throws InterruptedException {
        Thread thread1 = new Thread(() -> {
            for (int i = 0; i < 20; i++) {
                try {
                    System.out.println(i + "预备入队");
                    Thread.sleep(100);
                    arr.put(i);
                    System.out.println(i + "入队成功");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        thread1.start();
        Thread thread = new Thread(() -> {
            try {
                for (; ; ) {
                    System.out.println("进入消费队列");
                    System.out.println(arr.take());
                    Thread.sleep(1000);
                }
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
        });
        thread.start();
    }
}
  1. 开启了两个线程，一个是提供数据的线程，一个是消费数据的线程
  2. 入队操作之前进行了睡眠，目的是先让消费线程进行消费队列，然后队列数据提供线程再往线程中提供数据。
  3. 出队的操作中添加了sleep方法，目的是为了能让入队的内容多一些。
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
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
进入消费队列
0预备入队
0入队成功
1预备入队
1入队成功
2预备入队
0
2入队成功
3预备入队
3入队成功
4预备入队
4入队成功
5预备入队
5入队成功
6预备入队
6入队成功
7预备入队
7入队成功
8预备入队
8入队成功
9预备入队
9入队成功
10预备入队
10入队成功
11预备入队
进入消费队列
1
11入队成功
12预备入队
进入消费队列
2
12入队成功
13预备入队
进入消费队列
3
13入队成功
14预备入队
进入消费队列
4
14入队成功
15预备入队
进入消费队列
5
15入队成功
16预备入队
进入消费队列
6
16入队成功
17预备入队
进入消费队列
7
17入队成功
18预备入队
进入消费队列
8
18入队成功
19预备入队
进入消费队列
9
19入队成功
进入消费队列
10
进入消费队列
11
进入消费队列
12
进入消费队列
13
进入消费队列
14
进入消费队列
15
进入消费队列
16
进入消费队列
17
进入消费队列
18
进入消费队列
19
进入消费队列

分析上述输出内容：

进入消费队列文字输出出来说明数据提供者线程在休眠状态，而消费者线程在执行任务，在等待100ms后，0入队成功，1入队成功，当2准备入队时，这时候消费者线程获得了锁，消费了队列中的0,以此类推，最有一个进入消费队列说明队列为空等待队列不为空时，take方法进行消费。

通过输出结果可以得出以下内容：

  1. 消费线程先进入，但是并没有执行完，也就是说消费线程一直等待的状态。
  2. 入队和出队只能同步进行一项，也就是入队操作会阻止出队操作，出队操作也会阻止入队操作。

内部原理解析

ArrayBlockingQueue内部定义了以下的字段信息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
/** 队列元素数组 */
final Object[] items;

/** 下一个被take，poll，peek，remove的元素位置 */
int takeIndex;

/** 插入位置包含put，offer，add */
int putIndex;

/** 队列元素的数量 */
int count;

/** 重入锁 */
final ReentrantLock lock;

/** 队列不为空的条件 */
private final Condition notEmpty;

/** 队列满时的条件 */
private final Condition notFull;

当执行take()操作的时候，如果队列为空，则在notEmpty出等待，同时也会进行notFull的通知，通知notFull队列已经有位置可以进行入队操作了。新元素入队时，调用put方法时，如果队列满了，则当前线程暂停在notFull上，同时会进行一次notEmpty的通知，通知notEmpty队列已经有内容可，可以进行下面的内容了。

put方法源码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
/**
 * 将元素插入到队尾, 并且当队列中满的进行等待操作。
 *
 * @throws InterruptedException {@inheritDoc}
 * @throws NullPointerException {@inheritDoc}
 */
public void put(E e) throws InterruptedException {
    checkNotNull(e);
  	//获得重入锁。
    final ReentrantLock lock = this.lock;
  	//这里可以相应中断操作。
    lock.lockInterruptibly();
    try {
      	//判断队列是否已经满了。
        while (count == items.length)
          	//如果满了就在这里等待，等待通知队列为空时，进行相应。
            notFull.await();
      	//入队操作。
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

通过上面源代码可以得到以下内容：

  1. 获得重入锁进行入队同步操作，这也说明入队和出队只能同时进行一种操作的原因。
  2. 判断队列元素是否已经达到了队列的长度，也就是队列是否已经装满，如果装满则进行等待队列中有空余位置为止。
  3. 入队操作。

接下来详细看一下入队操作是如何进行的,enqueue源码如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
/**
 * 在当前putIndex位置插入元素, 并且通知notEmpty队列已经有内容.
 * 只有在获得锁的情况下执行.
 */
private void enqueue(E x) {
    // assert lock.getHoldCount() == 1;
    // assert items[putIndex] == null;
  	// 队列数组。
    final Object[] items = this.items;
  	// 在putIndex的位置插入x。
    items[putIndex] = x;
  	// 进行循环操作，如果putIndex到了队尾则将putIndex索引指向队头。
    if (++putIndex == items.length)
        putIndex = 0;
  	// 队列数量加1
    count++;
  	// 通知notEmpty队列已经有内容可以进行消费。
    notEmpty.signal();
}

接下来看一下take方法的源码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
/**
 * take方法从队列中获取元素，并且如果队列为空时进行notEmpty等待。
 */
public E take() throws InterruptedException {
    // 获得重入锁。
    final ReentrantLock lock = this.lock;
  	// 响应中断请求。
    lock.lockInterruptibly();
    try {
      	// 如果队列为空，则进行notEmpty等待。
        while (count == 0)
            notEmpty.await();
      	// 出队操作。
        return dequeue();
    } finally {
        lock.unlock();
    }
}

通过上面源码我们也可以看到如下内容：

  1. 出队前必须先获得锁，才能进行操作。
  2. 队列为空时，进行notEmpty等待。

dequeue方法源码如下所示：

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
/**
 * 出takeIndex元素信息,并且通知notFull队列已经有空余位置。
 * 执行必须先获得锁。
 */
private E dequeue() {
    // assert lock.getHoldCount() == 1;
    // assert items[takeIndex] != null;
  	// 队列元素。
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
  	// 获取takeIndex索引信息。
    E x = (E) items[takeIndex];
  	// 将takeIndex标记为null，方便GC进行回收。
    items[takeIndex] = null;
  	// 循环进行操作。
    if (++takeIndex == items.length)
        takeIndex = 0;
  	// 队列元素数量减1
    count--;
  	// 迭代器进行take
    if (itrs != null)
        itrs.elementDequeued();
  	// 通知notFull队列已经不再满，可进行put操作。
    notFull.signal();
    return x;
}

总结

通过对源码put和take 的分析总结一下几点：

  1. ArrayBlockingQueue是采用数组进行实现队列，通过putIndex和takeIndex来控制队列的队头和队尾。
  2. 内部使用ReentrantLock进行同步操作，并配合Condition处理等待操作。
  3. 总结成下面的图片内容：



下面来用图示法讲解下ArrayBlockingQueue的工作原理

首先将元素1进行入队操作，如下图所示：



putIndex和takeIndex在同一个位置，因为他只有一个元素，当再依次入队8个元素后内容如下所示

此时putIndex就到了最后一个数组的元素的索引上，当再向数组元素中添加元素时，就会进行notFull的等待操作

当调用take方法后，队列中出现了空余位置，并且通知了notFull，嘿，伙计你可以将你的东西添加到队列中了。

可以清晰的看到putIndex变到了数组索引0的位置，就是下面的代码导致的：

1
2
if (++putIndex == items.length)
    putIndex = 0;

并且此时的takeIndex便到了数组索引1的位置，持续进行take方法，队列内容全部出队列：

当take方法走到数组的末尾时，它会将takeIndex值设置为0，进行从新开始take。

1
2
if (++takeIndex == items.length)
    takeIndex = 0;

并且当队列为空时，会进行notEmpty等待，等待队列中存在元素，当调用put方法后，它会通知notEmpty，兄弟你可以取队列消息了。

本文标题:ArrayBlockingQueue原理详解

文章作者:BattleHeart

发布时间:2019-04-06, 14:16:31

字数统计: 2.5k字

阅读时长:9分

原始链接:https://www.battleheart.cn/2019/04/06/arrayblockingqueue-principle/

许可协议: "署名-非商用-相同方式共享 4.0" 转载请保留原文链接及作者。

文章目录
  1. 1. ArrayBlockingQueue原理详解
    1. 1.1. 介绍
      1. 1.1.1. 插入队列方法
      2. 1.1.2. 获取队列内容
    2. 1.2. 简单实例
    3. 1.3. 内部原理解析
    4. 1.4. 总结
|
载入天数...载入时分秒...