Jdk1.6 JUC源码解析(22)-LinkedBlockingDeque

Jdk1.6 JUC源码解析(22)-LinkedBlockingDeque

作者：大飞

 

功能简介：

• LinkedBlockingDeque是一种基于双向链表实现的有界的(可选的，不指定默认int最大值)阻塞双端队列。

       双端队列一般适用于工作密取模式，即每个消费者都拥有自己的双端队列，如果某个消费者完成了自己队列的全部任务，可以到其他消费者双端队列尾部秘密获取任务来处理。

 

源码分析：

• LinkedBlockingDeque实现了BlockingDeque接口，简单看下这个接口：

 

public interface BlockingDeque<E> extends BlockingQueue<E>, Deque<E> {
    /**
     * 将元素插入队头，如果队列满了，抛出IllegalStateException。
     */
    void addFirst(E e);
    /**
     * 将元素插入队尾，如果队列满了，抛出IllegalStateException。
     */
    void addLast(E e);
    /**
     * 将元素插入队头，如果成功，返回true；如果队列满了，返回false。
     */
    boolean offerFirst(E e);
    /**
     * 将元素插入队尾，如果成功，返回true；如果队列满了，返回false。
     */
    boolean offerLast(E e);
    /**
     * 将元素插入队头。如果队列满了，阻塞等待，直到队列有可用空间。
     */
    void putFirst(E e) throws InterruptedException;
    /**
     * 将元素插入队尾。如果队列满了，阻塞等待，直到队列有可用空间。
     */
    void putLast(E e) throws InterruptedException;
    /**
     * 将元素插入队头。如果队列满了，阻塞等待。
     * 如果插入成功，返回true；如果等待超时，返回false。
     */
    boolean offerFirst(E e, long timeout, TimeUnit unit)
        throws InterruptedException;
    /**
     * 将元素插入队尾。如果队列满了，阻塞等待。
     * 如果插入成功，返回true；如果等待超时，返回false。
     */
    boolean offerLast(E e, long timeout, TimeUnit unit)
        throws InterruptedException;
    /**
     * 获取并移除队头元素，如果队头没有元素，阻塞等待，直到有元素可以获取。
     */
    E takeFirst() throws InterruptedException;
    /**
     * 获取并移除队尾元素，如果队头没有元素，阻塞等待，直到有元素可以获取。
     */
    E takeLast() throws InterruptedException;
    /**
     * 获取并移除队头元素，如果队头没有元素，阻塞等待。
     * 如果过了给定的时间还不能获取元素，返回null。
     */
    E pollFirst(long timeout, TimeUnit unit)
        throws InterruptedException;
    /**
     * 获取并移除队尾元素，如果队头没有元素，阻塞等待。
     * 如果过了给定的时间还不能获取元素，返回null。
     */
    E pollLast(long timeout, TimeUnit unit)
        throws InterruptedException;
    /**
     * 从队列中删除第一个出现的指定元素。
     * 如果有这个元素并删除，返回true；如果没有指定元素，返回false。
     */
    boolean removeFirstOccurrence(Object o);
    /**
     * 从队列中删除最后一个出现的指定元素。
     * 如果有这个元素并删除，返回true；如果没有指定元素，返回false。
     */
    boolean removeLastOccurrence(Object o);
    // *** BlockingQueue methods ***
    /**
     * 添加一个元素到队尾，如果成功，返回true。如果队列满了，抛出IllegalStateException。
     */
    boolean add(E e);
    /**
     * 添加一个元素到队尾。添加成功，返回treue；如果队列满了，返回false。
     */
    boolean offer(E e);
    /**
     * 将元素插入队尾。如果队列满了，阻塞等待。
     */
    void put(E e) throws InterruptedException;
    /**
     * 将元素插入队尾。如果队列满了，阻塞等待。
     * 如果插入成功，返回true；如果等待超时，返回false。
     */
    boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException;
    /**
     * 获取并移除队头元素，如果队头没有元素，抛出NoSuchElementException异常。
     */
    E remove();
    /**
     * 获取并移除队头元素，如果队头没有元素，返回null。
     */
    E poll();
    /**
     * 获取并移除队头元素，如果队头没有元素，阻塞等待，直到有元素可以获取。
     */
    E take() throws InterruptedException;
    /**
     * 获取并移除队头元素，如果队头没有元素，阻塞等待。
     * 如果过了给定的时间还不能获取元素，返回null。
     */
    E poll(long timeout, TimeUnit unit)
        throws InterruptedException;
    /**
     * 查看队头元素，如果队头没有元素，抛出NoSuchElementException。
     */
    E element();
    /**
     * 查看队头元素，如果队头没有元素，返回null。
     */
    E peek();
    /**
     * 从队列中删除第一个出现的指定元素。
     * 如果有这个元素并删除，返回true；如果没有指定元素，返回false。
     */
    boolean remove(Object o);
    /**
     * 查看队列是否包含给定的元素，包含返回true；不包含返回false。
     */
    public boolean contains(Object o);
    /**
     * 返回队列中元素数量。
     */
    public int size();
    /**
     * 返回一个从头到尾排序的迭代器。
     */
    Iterator<E> iterator();
    // *** Stack methods ***
    /**
     * 相当于addFirst(Object)
     */
    void push(E e);
}

 

 

 

• 接下来看下LinkedBlockingDeque内部数据结构：

 

public class LinkedBlockingDeque<E>
    extends AbstractQueue<E>
    implements BlockingDeque<E>,  java.io.Serializable {
    private static final long serialVersionUID = -387911632671998426L;
    /** Doubly-linked list node class */
    static final class Node<E> {
       /**
        * 保存元素的域，如果为nul说明当前节点已被删除。
        */
	E item;
        /**
         * - 指向其前驱节点。
         * - 如果指向自身，说明前面是队尾节点。
         * - 如果为null，说明没有前驱节点。
         */
        Node<E> prev;
        /**
         * - 指向其后继节点。
         * - 如果指向自身，说明后面是队头节点。
         * - 如果为null，说明没有后继节点。
         */
        Node<E> next;
        Node(E x, Node<E> p, Node<E> n) {
            item = x;
            prev = p;
            next = n;
        }
    }
    /** 指向队头节点 */
    transient Node<E> first;
    /** 指向队尾节点 */
    transient Node<E> last;
    /** 队列中元素数量 */
    private transient int count;
    /** 队列最大容量 */
    private final int capacity;
    /** 队列中保护访问使用的锁 */
    final ReentrantLock lock = new ReentrantLock();
    /** 获取元素的等待条件(队列非空) */
    private final Condition notEmpty = lock.newCondition();
    /** 插入元素的等待条件(队列非满) */
    private final Condition notFull = lock.newCondition();
    /**
     * 不指定容量，默认为Integer.MAX_VALUE(可认为无限)。
     */
    public LinkedBlockingDeque() {
        this(Integer.MAX_VALUE);
    }

    public LinkedBlockingDeque(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
    }

       内部结构：一个双向链表、一把锁和两个锁条件。

 

• 再看下内部的一些基础操作，这些操作必须在持有锁的前提下调用：

    /**
     * Links e as first element, or returns false if full.
     */
    private boolean linkFirst(E e) {
        // assert lock.isHeldByCurrentThread();
        if (count >= capacity)
            return false; //如果队列已满，返回false；
        Node<E> f = first;
        //新建节点x，用来存放数据e，将e插入到队头节点前面。
        Node<E> x = new Node<E>(e, null, f);
        //然后将e设置为队头节点。 
        first = x;
        if (last == null)
            last = x; //如果没有队尾节点，那么将x设置为队尾节点。
        else
            f.prev = x; //如果有队尾节点，那么将f的prev指向x，完成节点拼接。
        ++count; // 累加当前元素计数
        notEmpty.signal(); //有元素入队了，唤醒在notEmpty上等待的获取元素的线程。
        return true;
    }

       linkFirst就是将一个元素插入到队头，并成为新的队头元素。 

 

    /**
     * Links e as last element, or returns false if full.
     */
    private boolean linkLast(E e) {
        // assert lock.isHeldByCurrentThread();
        if (count >= capacity)
            return false; //如果队列已满，返回false；
        Node<E> l = last;
        //新建节点x，用来存放数据e，将e插入到队尾节点后面。
        Node<E> x = new Node<E>(e, l, null);
        last = x;
        if (first == null)
            first = x; //如果没有队头节点，那么将x设置为队头节点。
        else
            l.next = x; //如果有队头节点，那么将l的next指向x，完成节点拼接。
        ++count; // 累加当前元素计数
        notEmpty.signal(); //有元素入队了，唤醒在notEmpty上等待的获取元素的线程。
        return true;
    }

       linkLast就是将一个元素插入到队尾，并成为新的队尾元素。 

 

    /**
     * Removes and returns first element, or null if empty.
     */
    private E unlinkFirst() {
        // assert lock.isHeldByCurrentThread();
        Node<E> f = first; //获取队头节点f。
        if (f == null)
            return null; //如果没有队头节点，返回null。
        Node<E> n = f.next; //获取f的后继节点。
        E item = f.item; //获取f的元素item。
        f.item = null; //将f的item域置空。
        f.next = f; // 将f的next域指向自身，帮助GC。
        first = n; //将f的后继节点n设置为新的队头节点。
        if (n == null)
            last = null; //如果n为空，说明队列为空了，把队尾节点也置空一下。
        else
            n.prev = null; //如果n不为空，现在n是队头节点，需要将其prev域置空。
        --count; //递减当前元素计数
        notFull.signal(); //有元素出队了，唤醒在notFull上等待的插入元素的线程。
        return item; //返回元素item。
    }

       unlinkFirst就是将现有的队头节点移除，并将其后继节点设置为新的队头节点，并返回移除的队头节点中保存的元素。 

 

    /**
     * Removes and returns last element, or null if empty.
     */
    private E unlinkLast() {
        // assert lock.isHeldByCurrentThread();
        Node<E> l = last; //获取队尾节点l。
        if (l == null)
            return null; //如果没有队尾节点，返回null。
        Node<E> p = l.prev; //获取l的前驱节点。
        E item = l.item; //获取l的元素item。
        l.item = null; //将l的item域置空。
        l.prev = l; // 将f的prev域指向自身，帮助GC。
        last = p; //将l的前驱节点p设置为新的队尾节点。
        if (p == null)
            first = null; //如果p为空，说明队列为空了，把队头节点也置空一下。
        else
            p.next = null; //如果p不为空，现在p是队尾节点，需要将其next域置空。
        --count; //递减当前元素计数
        notFull.signal(); //有元素出队了，唤醒在notFull上等待的插入元素的线程。
        return item; //返回元素item。
    }

      unlinkLast就是将现有的队尾节点移除，并将其前驱节点设置为新的队尾节点，并返回移除的队尾节点中保存的元素。 

 

    /**
     * Unlinks x
     */
    void unlink(Node<E> x) {
        // assert lock.isHeldByCurrentThread();
        Node<E> p = x.prev;
        Node<E> n = x.next;
        if (p == null) {
           unlinkFirst(); //如果x没有前驱节点，那么x就是队头节点，所以调用一下unlinkFirst就可以了。
        } else if (n == null) {
          unlinkLast(); //如果x没有后继节点，那么x就是队尾节点，所以调用一下unlinkLast就可以了。
        } else {
            p.next = n; //将x的前驱节点p的next指向x的后继节点n。
            n.prev = p; //将x的后继节点n的prev指向x的前驱节点n。
            x.item = null; //置空x的item域。
            // 注意上面并没有清理x本身的prev和next域，因为它们可能正在被某个迭代器使用中。
            --count; //递减当前元素计数
            notFull.signal(); //有元素出队了，唤醒在notFull上等待的插入元素的线程。
        }
    }

 

 

• LinkedBlockingDeque中利用上述基础方法来实现BlockingDeque接口定义的方法，随便看几个：

 

    public void putFirst(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            while (!linkFirst(e))
                notFull.await(); //如果插入元素到队头失败，在notFull条件上等待。
        } finally {
            lock.unlock();
        }
    }

    public void putLast(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            while (!linkLast(e))
                notFull.await(); //如果插入元素到队尾失败，在notFull条件上等待。
        } finally {
            lock.unlock();
        }
    }
    public E takeFirst() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            E x;
            while ( (x = unlinkFirst()) == null)
                notEmpty.await(); //如果从队头获取并删除元素失败，在notEmpty条件上等待。
            return x;
        } finally {
            lock.unlock();
        }
    }
    public E takeLast() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            E x;
            while ( (x = unlinkLast()) == null)
                notEmpty.await(); //如果从队尾获取并删除元素失败，在notEmpty条件上等待。
            return x;
        } finally {
            lock.unlock();
        }
    }

 

 

• 其他的方法也很容易看懂了，这里就不啰嗦了。最后注意LinkedBlockingDeque的迭代器是弱一致的，而且支持双向迭代器。

 

       LinkedBlockingDeque的代码解析完毕！

 

       参见：Jdk1.6 JUC源码解析(7)-locks-ReentrantLock

                  Jdk1.6 Collections Framework源码解析(3)-ArrayDeque