LinkedBlockingQueue源码解析

概念

LinkedBlockingQueue是非固定容量队列，内部是通过链表存放数据，并通过两把互斥锁（分别为取时互斥锁和放时互斥锁，这也是与ArrayBlockingQueue的本质区别，这样可以提高性能）来控制访问数据的同步问题。提供的方法与ArrayBlockingQueue基本是一致的，下面只介绍下有差异的地方，细节上看ArrayBlockingQueue的源码解读就行。
核心属性

/**
 * 阻塞队列元素封装为Node
 */
static class Node<E> {
    E item;

    Node<E> next;

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

/** 指定队列的长度，默认为Integer.MAX */
private final int capacity;

/** 记录数据条数 */
private final AtomicInteger count = new AtomicInteger();

//指向链表头
transient Node<E> head;
//指向链表尾
private transient Node<E> last;

/** 读锁 */
private final ReentrantLock takeLock = new ReentrantLock();
private final Condition notEmpty = takeLock.newCondition();

/** 写锁 */
private final ReentrantLock putLock = new ReentrantLock();
private final Condition notFull = putLock.newCondition();
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

2.1 写操作

// 写操作
public boolean offer(E e) {
    // 非空
    if (e == null) throw new NullPointerException();
    // 拿到count（记录当前数据条数）
    final AtomicInteger count = this.count;
    // 如果count达到了最大值
    if (count.get() == capacity)
        // 链表满了，添加失败，直接返回。
        return false;
    // 声明
    int c = -1;
    // 用当前元素创建一个新的Node
    Node<E> node = new Node<E>(e);
    // 写锁
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        // 再次判断链表是否还有空间，有则通过enqueue存入数据
        if (count.get() < capacity) {
            // 数据存入操作
            enqueue(node);
            // 链表元素数量count加1
            c = count.getAndIncrement();
            // 添加完数据之后，长度依然小于最大长度，链表还有存放空间，唤醒其他阻塞的写线程  
            // 读写不互斥，可能前面在执行时，队列是满的，但是读操作依然在进行
            if (c + 1 < capacity)
                notFull.signal();
        }
    } finally {
        putLock.unlock();  //解写锁
    }
    // 判断原先是否为空，c为0，表示原先链表为空，读线程肯定都在阻塞，所以，此时需求唤醒阻塞的读线程
    if (c == 0)
        signalNotEmpty();
    return c >= 0;
}

// 插入数据到链表（尾插法）
private void enqueue(Node<E> node) {
    last = last.next = node;
}
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

2.2 读操作

public E poll() {
    final AtomicInteger count = this.count;
    // 为0，没数据，退出
    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();
            // count --
            c = count.getAndDecrement();
            if (c > 1)
                // c > 1,说明还有数据，唤醒其他阻塞的读线程
                notEmpty.signal();
        }
    } finally {
        takeLock.unlock();
    }
    if (c == capacity)
        // 到这说明链接有空位，唤醒写线程
        signalNotFull();
    return x;
}
// 从头部取数据
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;
}
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