【JUC源码专题】LinkedBlockingDeque 源码分析（JDK8）

1
2
3
4
5
6
7
8
9
10
11
12
import java.util.concurrent.LinkedBlockingDeque;

public class TestQueue {
    public static void main(String[] args) throws InterruptedException {
        LinkedBlockingDeque<Integer> deque = new LinkedBlockingDeque<>(8);
        deque.putFirst(1);
        deque.getFirst();
        deque.putLast(2);
        deque.getLast();
    }
}

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
    static final class Node<E> {
        E item;
        Node<E> prev;
        Node<E> next;
        Node(E x) {
            item = x;
        }
    }
    
    // 链表头结点
    transient Node<E> first;

    // 链表尾结点
    transient Node<E> last;

    // 队列中元素的个数
    private transient int count;

    // 队列最大容量
    private final int capacity;

    // 全局锁
    final ReentrantLock lock = new ReentrantLock();

    // 用于等待队列为非空的条件变量（take 时队列空了，就要等到队列为非空时再进行 take）
    private final Condition notEmpty = lock.newCondition();

    // 用于等待队列为非满的条件变量（put 时队列满了，就要等到队列为非满时再进行 put）
    private final Condition notFull = lock.newCondition();

1
2
3
4
5
    public LinkedBlockingDeque(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
    public void putFirst(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        Node<E> node = new Node<E>(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            // 插入失败则等待在非满的条件变量上，直到插入成功
            while (!linkFirst(node))
                notFull.await();
        } finally {
            lock.unlock();
        }
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
    private boolean linkFirst(Node<E> node) {
        // 队列已满
        if (count >= capacity)
            return false;
        Node<E> f = first;
        // 将 node 的 next 指向当前头结点
        node.next = f;
        // 将 node 设置为新的头结点
        first = node;
        // 若尾结点为空，将 node 设置为尾结点
        if (last == null)
            last = node;
        else
            f.prev = node; // 当前头结点的 prev 指向 node
        // 队列元素个数+1
        ++count;
        // 唤醒等待队列非空的线程，即唤醒消费者
        notEmpty.signal();
        return true;
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
    public void putLast(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        Node<E> node = new Node<E>(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            while (!linkLast(node))
                notFull.await();
        } finally {
            lock.unlock();
        }
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
    private boolean linkLast(Node<E> node) {
        if (count >= capacity)
            return false;
        Node<E> l = last;
        // 将 node 的 prev 指向当前的 last 结点
        node.prev = l;
        // 将 node 设置为新的 last 结点
        last = node;
        // 若头结点为空，则头结点也指向 node
        if (first == null)
            first = node;
        else
            l.next = node; // 否则将当前 last 结点的 next 指向 node
        // 队列元素数量加一
        ++count;
        // 唤醒消费者
        notEmpty.signal();
        return true;
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
    public E takeFirst() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            E x;
            while ( (x = unlinkFirst()) == null)
                notEmpty.await();
            return x;
        } 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
    private E unlinkFirst() {
        Node<E> f = first;
        if (f == null)
            return null;
        // 获取头结点的 next 结点 n
        Node<E> n = f.next;
        // 获取头结点的数据
        E item = f.item;
        // 清空对数据的引用
        f.item = null;
        // f 的 next 指向自己，帮助 gc
        f.next = f; // help GC
        // 更新头结点
        first = n;
        // 若 n 为空，则说明此时队列为空，last 指向 null
        if (n == null)
            last = null;
        else
            n.prev = null; // n 的 prev 设置为 null
        // 队列元素数量减一
        --count;
        // 唤醒等待队列为非满的线程，即唤醒生产者
        notFull.signal();
        return item;
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
    public E takeLast() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            E x;
            // 若队列为空，则线程等待在非空条件上
            while ( (x = unlinkLast()) == null)
                notEmpty.await();
            return x;
        } 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
    private E unlinkLast() {
        Node<E> l = last;
        if (l == null)
            return null;
        // 获取尾结点的 prev 结点 p
        Node<E> p = l.prev;
        // 获取当前尾结点的 item
        E item = l.item;
        // 清空当前尾结点对数据的引用
        l.item = null;
        // 当前尾结点的 prev 指向自己
        l.prev = l; // help GC
        // 更新 p 为新的尾结点
        last = p;
        // 若 p 等于 null，则说明此时链表 unlinkLast 之前就只有一个结点，更新头结点为 null
        if (p == null)
            first = null;
        else
            p.next = null; // 新的尾结点 p 的 next 设置为 null
        // 队列元素数量减一
        --count;
        // 唤醒等待队列非满的线程，即唤醒生产者
        notFull.signal();
        return item;
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
    public boolean contains(Object o) {
        if (o == null) return false;
        final ReentrantLock lock = this.lock;
        // 获取锁
        lock.lock();
        try {
            // 遍历
            for (Node<E> p = first; p != null; p = p.next)
                if (o.equals(p.item))
                    return true;
            return false;
        } finally {
            lock.unlock();
        }
    }