Java并发编程与技术内幕:ArrayBlockingQueue、LinkedBlockingQueue及SynchronousQueue源码解析

本是古典何须时尚 2021-09-11 00:14 325阅读 0赞

林炳文Evankaka原创作品。转载请注明出处[http://blog.csdn.net/evankaka][http_blog.csdn.net_evankaka]

摘要：本文主要讲了Java中BlockingQueue的源码

# 一、BlockingQueue介绍与常用方法 #

BlockingQueue是一个阻塞队列。在高并发场景是用得非常多的，在线程池中。如果运行线程数目大于核心线程数目时，也会尝试把新加入的线程放到一个BlockingQueue中去。队列的特性就是先进先出很容易理解，在java里头它的实现类主要有下图的几种，其中最常用到的是ArrayBlockingQueue、LinkedBlockingQueue及SynchronousQueue这三种，这三个也是今天主要讲的类。

![20160618160251536][]

它主要的方法有

![20160618155259551][]

BlockingQueue的核心方法：  
1、放入数据

（1） add(object)

队列没满的话，放入成功。否则抛出异常。

（2）offer(object):

表示如果可能的话,将object加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则返回false.（本方法不阻塞当前执行方法的线程）  
 （3）offer(E o, long timeout, TimeUnit unit)

可以设定等待的时间，如果在指定的时间内，还不能往队列中加入BlockingQueue，则返回失败。  
（4）put(object)

把object加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程阻塞。直到BlockingQueue里面有空间再继续.  
2、获取数据  
（1）poll(time)

取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回null;  
（2）poll(long timeout, TimeUnit unit)

从BlockingQueue取出一个队首的对象，如果在指定时间内，队列一旦有数据可取，则立即返回队列中的数据。否则知道时间超时还没有数据可取，返回失败。

（3）take()

取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的数据被加入;  
（4）drainTo()

一次性从BlockingQueue获取所有可用的数据对象（还可以指定获取数据的个数），通过该方法，可以提升获取数据效率；不需要多次分批加锁或释放锁。

# 二、ArrayBlockingQueue #

一个基本数组的阻塞队列。可以设置列队的大小。

ArrayBlockingQueue的源码是比较简单的，下面是笔者抽取了一部分源码并加以注释。它的基本原理实际还是数组，只不过存、取、删时都要做队列是否满或空的判断。然后加锁访问。

package java.util.concurrent;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.ReentrantLock;
    import java.util.AbstractQueue;
    import java.util.Collection;
    import java.util.Iterator;
    import java.util.NoSuchElementException;
    import java.lang.ref.WeakReference;
    import java.util.Spliterators;
    import java.util.Spliterator;
    
    
    public class ArrayBlockingQueue<E> extends AbstractQueue<E>
            implements BlockingQueue<E>, java.io.Serializable {
    
        private static final long serialVersionUID = -817911632652898426L;
    
        /** 真正存入数据的数组*/
        final Object[] items;
    
        /** take, poll, peek or remove的下一个索引 */
        int takeIndex;
    
        /** put, offer, or add的下一个索引 */
        int putIndex;
    
        /**队列中元素个数*/
        int count;
    
    
        /**可重入锁 */
        final ReentrantLock lock;
    
        /** 队列不为空的条件 */
        private final Condition notEmpty;
    
        /** 队列未满的条件 */
        private final Condition notFull;
    
        transient Itrs itrs = null;
    
    
        /**
         *当前元素个数-1
         */
        final int dec(int i) {
            return ((i == 0) ? items.length : i) - 1;
        }
    
        /**
         * 返回对应索引上的元素
         */
        @SuppressWarnings("unchecked")
        final E itemAt(int i) {
            return (E) items[i];
        }
    
        /**
         * 非空检查
         *
         * @param v the element
         */
        private static void checkNotNull(Object v) {
            if (v == null)
                throw new NullPointerException();
        }
    
        /**
         * 元素放入队列，注意调用这个方法时都要先加锁
         * 
         */
        private void enqueue(E x) {
            final Object[] items = this.items;
            items[putIndex] = x;
            if (++putIndex == items.length)
                putIndex = 0;
            count++;//当前拥有元素个数加1
            notEmpty.signal();//有一个元素加入成功，那肯定队列不为空
        }
    
        /**
         * 元素出队，注意调用这个方法时都要先加锁
         * 
         */
        private E dequeue() {
            final Object[] items = this.items;
            @SuppressWarnings("unchecked")
            E x = (E) items[takeIndex];
            items[takeIndex] = null;
            if (++takeIndex == items.length)
                takeIndex = 0;
            count--;/当前拥有元素个数减1
            if (itrs != null)
                itrs.elementDequeued();
            notFull.signal();//有一个元素取出成功，那肯定队列不满
            return x;
        }
    
        /**
         * 指定删除索引上的元素
         * 
         */
        void removeAt(final int removeIndex) {
            final Object[] items = this.items;
            if (removeIndex == takeIndex) {
                items[takeIndex] = null;
                if (++takeIndex == items.length)
                    takeIndex = 0;
                count--;
                if (itrs != null)
                    itrs.elementDequeued();
            } else {
                final int putIndex = this.putIndex;
                for (int i = removeIndex;;) {
                    int next = i + 1;
                    if (next == items.length)
                        next = 0;
                    if (next != putIndex) {
                        items[i] = items[next];
                        i = next;
                    } else {
                        items[i] = null;
                        this.putIndex = i;
                        break;
                    }
                }
                count--;
                if (itrs != null)
                    itrs.removedAt(removeIndex);
            }
            notFull.signal();//有一个元素删除成功，那肯定队列不满
        }
    
        /**
         * 
         * 构造函数，设置队列的初始容量
         */
        public ArrayBlockingQueue(int capacity) {
            this(capacity, false);
        }
    
        /**
         * 构造函数。capacity设置数组大小 ，fair设置是否为公平锁
         * capacity and the specified access policy.
         */
        public ArrayBlockingQueue(int capacity, boolean fair) {
            if (capacity <= 0)
                throw new IllegalArgumentException();
            this.items = new Object[capacity];
            lock = new ReentrantLock(fair);//是否为公平锁，如果是的话，那么先到的线程先获得锁对象。
            //否则，由操作系统调度由哪个线程获得锁，一般为false，性能会比较高
            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(); //要给数组设置内容，先上锁
            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;//如果putIndex大于数组大小 ，那么从0重新开始
            } finally {
                lock.unlock();//最后一定要释放锁
            }
        }
    
        /**
         * 添加一个元素，其实super.add里面调用了offer方法
         */
        public boolean add(E e) {
            return super.add(e);
        }
    
        /**
         *加入成功返回true,否则返回false
         * 
         */
        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();
            }
        }
    
        /**
         * 如果队列已满的话，就会等待
         */
        public void put(E e) throws InterruptedException {
            checkNotNull(e);
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();//和lock()方法的区别是让它在阻塞时也可抛出异常跳出
            try {
                while (count == items.length)
                    notFull.await(); //这里就是阻塞了，要注意。如果运行到这里，那么它会释放上面的锁，一直等到notify
                enqueue(e);
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * 带有超时时间的插入方法，unit表示是按秒、分、时哪一种
         */
        public boolean offer(E e, long timeout, TimeUnit unit)
            throws InterruptedException {
    
            checkNotNull(e);
            long nanos = unit.toNanos(timeout);
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                while (count == items.length) {
                    if (nanos <= 0)
                        return false;
                    nanos = notFull.awaitNanos(nanos);//带有超时等待的阻塞方法
                }
                enqueue(e);//入队
                return true;
            } finally {
                lock.unlock();
            }
        }
    
        //实现的方法，如果当前队列为空，返回null
        public E poll() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return (count == 0) ? null : dequeue();
            } finally {
                lock.unlock();
            }
        }
         //实现的方法，如果当前队列为空，一直阻塞
        public E take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                while (count == 0)
                    notEmpty.await();//队列为空，阻塞方法
                return dequeue();
            } finally {
                lock.unlock();
            }
        }
        //带有超时时间的取元素方法，否则返回Null
        public E poll(long timeout, TimeUnit unit) throws InterruptedException {
            long nanos = unit.toNanos(timeout);
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                while (count == 0) {
                    if (nanos <= 0)
                        return null;
                    nanos = notEmpty.awaitNanos(nanos);//超时等待
                }
                return dequeue();//取得元素
            } finally {
                lock.unlock();
            }
        }
        //只是看一个队列最前面的元素，取出是不删除队列中的原来元素。队列为空时返回null
        public E peek() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return itemAt(takeIndex); // 队列为空时返回null
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * 返回队列当前元素个数
         *
         */
        public int size() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return count;
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * 返回当前队列再放入多少个元素就满队
         */
        public int remainingCapacity() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return items.length - count;
            } finally {
                lock.unlock();
            }
        }
    
        /**
         *  从队列中删除一个元素的方法。删除成功返回true,否则返回false
         */
        public boolean remove(Object o) {
            if (o == null) return false;
            final Object[] items = this.items;
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                if (count > 0) {
                    final int putIndex = this.putIndex;
                    int i = takeIndex;
                    do {
                        if (o.equals(items[i])) {
                            removeAt(i); //真正删除的方法
                            return true;
                        }
                        if (++i == items.length)
                            i = 0;
                    } while (i != putIndex);//一直不断的循环取出来做判断
                }
                return false;
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * 是否包含一个元素
         */
        public boolean contains(Object o) {
            if (o == null) return false;
            final Object[] items = this.items;
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                if (count > 0) {
                    final int putIndex = this.putIndex;
                    int i = takeIndex;
                    do {
                        if (o.equals(items[i]))
                            return true;
                        if (++i == items.length)
                            i = 0;
                    } while (i != putIndex);
                }
                return false;
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * 清空队列
         *
         */
        public void clear() {
            final Object[] items = this.items;
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                int k = count;
                if (k > 0) {
                    final int putIndex = this.putIndex;
                    int i = takeIndex;
                    do {
                        items[i] = null;
                        if (++i == items.length)
                            i = 0;
                    } while (i != putIndex);
                    takeIndex = putIndex;
                    count = 0;
                    if (itrs != null)
                        itrs.queueIsEmpty();
                    for (; k > 0 && lock.hasWaiters(notFull); k--)
                        notFull.signal();
                }
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * 取出所有元素到集合
         */
        public int drainTo(Collection<? super E> c) {
            return drainTo(c, Integer.MAX_VALUE);
        }
    
        /**
         * 取出所有元素到集合
         */
        public int drainTo(Collection<? super E> c, int maxElements) {
            checkNotNull(c);
            if (c == this)
                throw new IllegalArgumentException();
            if (maxElements <= 0)
                return 0;
            final Object[] items = this.items;
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                int n = Math.min(maxElements, count);
                int take = takeIndex;
                int i = 0;
                try {
                    while (i < n) {
                        @SuppressWarnings("unchecked")
                        E x = (E) items[take];
                        c.add(x);
                        items[take] = null;
                        if (++take == items.length)
                            take = 0;
                        i++;
                    }
                    return n;
                } finally {
                    // Restore invariants even if c.add() threw
                    if (i > 0) {
                        count -= i;
                        takeIndex = take;
                        if (itrs != null) {
                            if (count == 0)
                                itrs.queueIsEmpty();
                            else if (i > take)
                                itrs.takeIndexWrapped();
                        }
                        for (; i > 0 && lock.hasWaiters(notFull); i--)
                            notFull.signal();
                    }
                }
            } finally {
                lock.unlock();
            }
        }
    
    
    }

# 三、LinkedBlockingQueue #

接下来看看LinkedBlockingQueue的部分源码。

package java.util.concurrent;
    import java.util.concurrent.atomic.AtomicInteger;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.ReentrantLock;
    import java.util.AbstractQueue;
    import java.util.Collection;
    import java.util.Iterator;
    import java.util.NoSuchElementException;
    import java.util.Spliterator;
    import java.util.Spliterators;
    import java.util.function.Consumer;
    
    public class LinkedBlockingQueue<E> extends AbstractQueue<E>
            implements BlockingQueue<E>, java.io.Serializable {
        private static final long serialVersionUID = -6903933977591709194L;
    
    
    
        /**
         * 链表节点类
         */
        static class Node<E> {
            E item;
            Node<E> next;//下一节点
            Node(E x) { item = x; }
        }
    
        /** 链表大小 ，默认大小 是Integer.MAX_VALUE */
        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();
    
        /**
         * 不为空通知方法
         */
        private void signalNotEmpty() {
            final ReentrantLock takeLock = this.takeLock;
            takeLock.lock();
            try {
                notEmpty.signal();
            } finally {
                takeLock.unlock();
            }
        }
    
        /**
         * 不为满通知方法
         */
        private void signalNotFull() {
            final ReentrantLock putLock = this.putLock;
            putLock.lock();
            try {
                notFull.signal();
            } finally {
                putLock.unlock();
            }
        }
    
        /**
         * 进队
         *
         * @param node the node
         */
        private void enqueue(Node<E> node) {
            last = last.next = node;
        }
    
        /**
         * 出队
         */
        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;
        }
    
        /**
         * 取和入都上锁，此时无法取和放
         */
        void fullyLock() {
            putLock.lock();
            takeLock.lock();
        }
    
        /**
         * 释放锁
         */
        void fullyUnlock() {
            takeLock.unlock();
            putLock.unlock();
        }
    
    
        /**
         * 构造函数
         */
        public LinkedBlockingQueue() {
            this(Integer.MAX_VALUE);
        }
    
        /**
         * 构造函数
         *
         */
        public LinkedBlockingQueue(int capacity) {
            if (capacity <= 0) throw new IllegalArgumentException();
            this.capacity = capacity;
            last = head = new Node<E>(null);
        }
    
        /**
         * 构造函数
         */
        public LinkedBlockingQueue(Collection<? extends E> c) {
            this(Integer.MAX_VALUE);
            final ReentrantLock putLock = this.putLock;
            putLock.lock(); //取得放入锁
            try {
                int n = 0;
                for (E e : c) {
                    if (e == null)
                        throw new NullPointerException();
                    if (n == capacity)
                        throw new IllegalStateException("Queue full");
                    enqueue(new Node<E>(e));
                    ++n;
                }
                count.set(n);
            } finally {
                putLock.unlock();
            }
        }
    
    
         //阻塞等待放入
        public void put(E e) throws InterruptedException {
            if (e == null) throw new NullPointerException();
            int c = -1;
            Node<E> node = new Node<E>(e);
            final ReentrantLock putLock = this.putLock;
            final AtomicInteger count = this.count;
            putLock.lockInterruptibly(); //取得放入锁
            try {
                while (count.get() == capacity) {//队列已满
                    notFull.await();
                }
                enqueue(node);//入队
                c = count.getAndIncrement();//当前队列中元素个数加1
                if (c + 1 < capacity)
                    notFull.signal();
            } finally {
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty();
        }
    
        /**
         *带超时时间的阻塞等待放入，队列不满。放入成功返回true，否则返回fasle
         */
        public boolean offer(E e, long timeout, TimeUnit unit)
            throws InterruptedException {
    
            if (e == null) throw new NullPointerException();
            long nanos = unit.toNanos(timeout);
            int c = -1;
            final ReentrantLock putLock = this.putLock;
            final AtomicInteger count = this.count;
            putLock.lockInterruptibly();
            try {
                while (count.get() == capacity) {
                    if (nanos <= 0)
                        return false;
                    nanos = notFull.awaitNanos(nanos);
                }
                enqueue(new Node<E>(e));
                c = count.getAndIncrement();
                if (c + 1 < capacity)
                    notFull.signal();
            } finally {
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty();
            return true;
        }
    
        /**
         * 非阻塞放入。队列不满放入成功返回true，否则返回fasle
         */
        public boolean offer(E e) {
            if (e == null) throw new NullPointerException();
            final AtomicInteger count = this.count;
            if (count.get() == capacity)
                return false;
            int c = -1;
            Node<E> node = new Node<E>(e);
            final ReentrantLock putLock = this.putLock;
            putLock.lock();
            try {
                if (count.get() < capacity) {
                    enqueue(node);
                    c = count.getAndIncrement();
                    if (c + 1 < capacity)
                        notFull.signal();
                }
            } finally {
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty();
            return c >= 0;
        }
        //阻塞等待取出元素
        public E take() throws InterruptedException {
            E x;
            int c = -1;
            final AtomicInteger count = this.count;
            final ReentrantLock takeLock = this.takeLock;
            takeLock.lockInterruptibly();
            try {
                while (count.get() == 0) {
                    notEmpty.await();
                }
                x = dequeue();
                c = count.getAndDecrement();
                if (c > 1)
                    notEmpty.signal();
            } finally {
                takeLock.unlock();
            }
            if (c == capacity)
                signalNotFull();
            return x;
        }
        //带有超时时间等待的取出元素
        public E poll(long timeout, TimeUnit unit) throws InterruptedException {
            E x = null;
            int c = -1;
            long nanos = unit.toNanos(timeout);
            final AtomicInteger count = this.count;
            final ReentrantLock takeLock = this.takeLock;
            takeLock.lockInterruptibly();//等待时可抛出异常跳出
            try {
                while (count.get() == 0) {
                    if (nanos <= 0)
                        return null;
                    nanos = notEmpty.awaitNanos(nanos);//超时等待
                }
                x = dequeue();
                c = count.getAndDecrement();
                if (c > 1)
                    notEmpty.signal();//不这空条件成立
            } finally {
                takeLock.unlock();
            }
            if (c == capacity)
                signalNotFull();
            return x;
        }
        //取队头元素。没有的话返回null，有的话返回元素，并将队列中删除此元素
        public E poll() {
            final AtomicInteger count = this.count;
            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();//出队
                    c = count.getAndDecrement();//当前队列中元素个数减去1
                    if (c > 1)
                        notEmpty.signal();//不为空条件成功
                }
            } finally {
                takeLock.unlock();
            }
            if (c == capacity)
                signalNotFull();
            return x;
        }
        //取队头元素，但不从队列中删除 ，没有的话返回null,不阻塞
        public E peek() {
            if (count.get() == 0)
                return null;
            final ReentrantLock takeLock = this.takeLock;
            takeLock.lock();//获得取得锁
            try {
                Node<E> first = head.next;
                if (first == null)
                    return null;
                else
                    return first.item;
            } finally {
                takeLock.unlock();
            }
        }
    
     
        /**
         * 删除时要同时取得放入锁和取得锁
         */
        public boolean remove(Object o) {
            if (o == null) return false;
            fullyLock();//同时取得放入锁和取得锁
            try {
                for (Node<E> trail = head, p = trail.next;
                     p != null;
                     trail = p, p = p.next) {
                    if (o.equals(p.item)) {
                        unlink(p, trail);
                        return true;
                    }
                }
                return false;
            } finally {
                fullyUnlock();
            }
        }
    
        /**
         * 是否包含
         */
        public boolean contains(Object o) {
            if (o == null) return false;
            fullyLock();//同时取得放入锁和取得锁
            try {
                for (Node<E> p = head.next; p != null; p = p.next)
                    if (o.equals(p.item))
                        return true;
                return false;
            } finally {
                fullyUnlock();
            }
        }
    }

**从LinkedBlockingQueue的源码中，我们可以看出他和ArrayBlockingQueue主要有以下两点区别：**

**1、ArrayBlockingQueue数据是放在一个数组中。LinkedBlockingQueue是放在一个Node节点中，构成一个链接。**

**2、ArrayBlockingQueue取元素和放元素都是同一个锁，而LinkedBlockingQueue有两个锁，一个放入锁，一个取得锁。分别对应放入元素和取得元素时的操作。这是由链表的结构所确定的。但是删除一个元素时，要同时获得放入锁和取得锁。**

# 四、SynchronousQueue  #

SynchronousQueue 这个队列实现了 BlockingQueue接口。该队列的特点  
1.容量为0，无论何时 size方法总是返回0  
2. put操作阻塞， 直到另外一个线程取走队列的元素。  
3.take操作阻塞，直到另外的线程put某个元素到队列中。  
4. 任何线程只能取得其他线程put进去的元素，而不会取到自己put进去的元素

public SynchronousQueue(boolean fair) {
            transferer = fair ? new TransferQueue() : new TransferStack();
        }

构造方法上接收boolean参数，表示这是一个公平的基于队列的排队模式，还是一个非公平的基于栈的排队模式。

[http_blog.csdn.net_evankaka]: http://blog.csdn.net/evankaka
[20160618160251536]: /images/20210811/076b546349ec49d6a2c90ea9845ad71a.png
[20160618155259551]: /images/20210811/076440f4491345968d363e28a9f6a46a.png