Fork me on GitHub

【Java多线程】JUC集合 04. ArrayBlockingQueue

发表于 | 分类于 ，

ArrayBlockingQueue

1. 前言

  • 通过循环数组实现的线程安全的有界阻塞队列
  • FIFO队列
  • 内部通过互斥锁ReentrantLock来实现多线程对竞争资源的互斥访问

2. 源码分析

2.1 数据结构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {

    /** The queued items */
    final Object[] items;

    /** items index for next take, poll, peek or remove */
    int takeIndex; 	//下一个被取出元素的索引

    /** items index for next put, offer, or add */
    int putIndex;	//下一个被添加元素的索引

    /** Number of elements in the queue */
    int count;	//队列中元素个数

    /** Main lock guarding all access */
    final ReentrantLock lock;

    /** Condition for waiting takes */
    private final Condition notEmpty;

    /** Condition for waiting puts */
    private final Condition notFull;
}

UML类图如下：

2.2 核心函数

  • 构造函数
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
public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
}

//capacity: 数组容量, fail: true/false - 公平锁/非公平锁
public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)
        throw new IllegalArgumentException();
    this.items = new Object[capacity]; 
    lock = new ReentrantLock(fair); 
    notEmpty = lock.newCondition();//锁非空 条件
    notFull =  lock.newCondition();//锁满 条件
}

public ArrayBlockingQueue(int capacity, boolean fair,
                          Collection<? extends E> c) {
    this(capacity, fair);

    final ReentrantLock lock = this.lock;
    lock.lock(); // Lock only for visibility, not mutual exclusion
    try {
        int i = 0;
        try {
            for (E e : c) {
                checkNotNull(e);
                items[i++] = e;
            }
        } catch (ArrayIndexOutOfBoundsException ex) {
            throw new IllegalArgumentException();
        }
        count = i;
        putIndex = (i == capacity) ? 0 : i;//入数组指针
    } finally {
        lock.unlock();
    }
}
  • 添加元素：offer(E e) 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
//将e插入阻塞队列尾部，如果队列已满，返回false; 否则插入成功，返回true
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();
    }
}

实际调用enqueue(E x)方法：

1
2
3
4
5
6
7
8
9
10
private void enqueue(E x) {
    // assert lock.getHoldCount() == 1;
    // assert items[putIndex] == null;
    final Object[] items = this.items;
    items[putIndex] = x;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
    notEmpty.signal();//唤醒notEmpty上等待的线程
}
  • 取出元素： take()
1
2
3
4
5
6
7
8
9
10
11
12
//取出阻塞队列的头部元素并返回，若队列为空则等待
public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}

实际调用的dequeue()方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
private E dequeue() {
    // assert lock.getHoldCount() == 1;
    // assert items[takeIndex] != null;
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    notFull.signal();//唤醒notFull上等待的线程
    return x;
}

3. 参考

http://www.cnblogs.com/skywang12345/p/3498652.html