JDK1.8 ArrayBlockingQueue源码分析

水深无声 2022-07-17 02:52 238阅读 0赞

**目的：本文通过分析JDK源码来对比ArrayBlockingQueue 和LinkedBlockingQueue，以便日后灵活使用。**

1. 在Java的Concurrent包中，添加了阻塞队列BlockingQueue，用于多线程编程。BlockingQueue的核心方法有：

boolean add(E e) ，把 e 添加到BlockingQueue里。如果BlockingQueue可以容纳，则返回true，否则抛出异常。 
boolean offer(E e)，表示如果可能的话,将 e 加到BlockingQueue里，即如果BlockingQueue可以容纳，则返回true，否则返回false。 
void put(E e)，把 e 添加到BlockingQueue里，如果BlockQueue没有空间，则调用此方法的线程被阻塞直到BlockingQueue里面有空间再继续。 
E poll(long timeout, TimeUnit unit) ，取走BlockingQueue里排在首位的对象，若不能立即取出，则可以等time参数规定的时间，取不到时返回null。 
E take() ，取走BlockingQueue里排在首位的对象,若BlockingQueue为空，则调用此方法的线程被阻塞直到BlockingQueue有新的数据被加入。 
int drainTo(Collection<? super E> c) 和 int drainTo(Collection<? super E> c, int maxElements) ，一次性从BlockingQueue获取所有可用的数据对象（还可以指定获取　数据的个数），通过该方法，可以提升获取数据效率，不需要多次分批加锁或释放锁。

注意：BlockingQueue 不接受null 元素。试图add、put 或offer 一个null 元素时，某些实现会抛出NullPointerException。null 被用作指示poll 操作失败的警戒值。

2. BlockingQueue常用的四个实现类

ArrayBlockingQueue：规定大小的BlockingQueue,其构造函数必须带一个int参数来指明其大小.其所含的对象是以FIFO(先入先出)顺序排序的.  
2) LinkedBlockingQueue：大小不定的BlockingQueue,若其构造函数带一个规定大小的参数,生成的BlockingQueue有大小限制,若不带大小参数,所生成的BlockingQueue的大小由Integer.MAX\_VALUE来决定.其所含的对象是以FIFO(先入先出)顺序排序的  
3) PriorityBlockingQueue：类似于LinkedBlockQueue,但其所含对象的排序不是FIFO,而是依据对象的自然排序顺序或者是构造函数的Comparator决定的顺序.  
4) SynchronousQueue：特殊的BlockingQueue,对其的操作必须是放和取交替完成的.

本文将从JDK源码层面分析对比ArrayBlockingQueue和LinkedBlockingQueue

3. ArrayBlockingQueue源码分析

ArrayBlockingQueue是一个由数组支持的有界阻塞队列。此队列按 FIFO（先进先出）原则对元素进行排序。队列的头部 是在队列中存在时间最长的元素，队列的尾部 是在队列中存在时间最短的元素。新元素插入到队列的尾部，队列检索操作则是从队列头部开始获得元素。  
这是一个典型的“有界缓存区”，固定大小的数组在其中保持生产者插入的元素和使用者提取的元素。一旦创建了这样的缓存区，就不能再增加其容量。试图向已满队列中放入元素会导致放入操作受阻塞；试图从空队列中检索元素将导致类似阻塞。

ArrayBlockingQueue创建的时候需要指定容量capacity(可以存储的最大的元素个数，因为它不会自动扩容)。其中一个构造方法为：

/**
 * Creates an {@code ArrayBlockingQueue} with the given (fixed)
 * capacity and the specified access policy.
 *
 * @param capacity the capacity of this queue
 * @param fair if {@code true} then queue accesses for threads blocked
 * on insertion or removal, are processed in FIFO order;
 * if {@code false} the access order is unspecified.
 * @throws IllegalArgumentException if {@code capacity < 1}
 */
 //传入一个消息队列长度，是否是公平队列
 public ArrayBlockingQueue(int capacity, boolean fair) {
 if (capacity <= 0)
 throw new IllegalArgumentException();
 //new 出一个指定长度的消息队列，这里要注意一旦消息队列里面满的话再往里面发消息就堵塞不会像ArrayList一样扩容的，
 this.items = new Object[capacity];
 lock = new ReentrantLock(fair);
 //初始化等待条件，关于condition可以看下这篇文章http://blog.csdn.net/qq_22929803/article/details/52347115
 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);
 // 存入ArrayBlockingQueue中
 items[i++] = e;
 }
 } catch (ArrayIndexOutOfBoundsException ex) { // 当初始化容量小于传入集合的大小时，会抛出异常
 throw new IllegalArgumentException();
 }
 // 元素数量
 count = i;
 // 初始化存元素的索引
 putIndex = (i == capacity) ? 0 : i;
 } finally {
 // 释放锁
 lock.unlock();
 }
 }

ArrayBlockingQueue类中定义的变量有：

/** 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;
    
        /*
         * Concurrency control uses the classic two-condition algorithm
         * found in any textbook.
         */
    
        /** Main lock guarding all access */
    // 整个消息队列的锁
        final ReentrantLock lock;
    
        /** Condition for waiting takes */
    // 等待获取条件
        private final Condition notEmpty;
    
        /** Condition for waiting puts */
    // 等待存放条件
        private final Condition notFull;
    
        /**
         * Shared state for currently active iterators, or null if there
         * are known not to be any.  Allows queue operations to update
         * iterator state.
         */
    // 迭代器
        transient Itrs itrs = null;
    
        // Internal helper methods

使用数组items来存储元素，由于是循环队列，使用takeIndex和putIndex来标记put和take的位置。可以看到，该类中只定义了一个锁ReentrantLock，定义两个Condition对象：notEmputy和notFull，分别用来对take和put操作进行所控制。注：本文主要讲解put()和take()操作,其他方法类似。

put(E e)方法的源码如下。进行put操作之前，必须获得锁并进行加锁操作，以保证线程安全性。加锁后，若发现队列已满，则调用notFull.await()方法，如当前线程陷入等待。直到其他线程take走某个元素后，会调用notFull.signal()方法来激活该线程。激活之后，继续下面的插入操作。

public void put(E e) throws InterruptedException {
            checkNotNull(e);
            // 获取可重入锁
            final ReentrantLock lock = this.lock;
            // 如果当前线程未被中断，则获取锁
            lock.lockInterruptibly();
            try {
                while (count == items.length) // 判断元素是否已满
                    // 若满，则等待
                    notFull.await();
                // 入队列
                enqueue(e);
            } finally {
                // 释放锁
                lock.unlock();
            }
        }

说明：put函数用于存放元素，在当前线程被中断时会抛出异常，并且当队列已经满时，会阻塞一直等待。其中，put会调用enqueue函数，enqueue函数源码如下

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) // 放入后存元素的索引等于数组长度（表示已满）
                // 重置存索引为0
                putIndex = 0;
            // 元素数量加1
            count++;
            // 唤醒在notEmpty条件上等待的线程
            notEmpty.signal();
        }

说明：enqueue函数用于将元素存入底层Object数组中，并且会唤醒等待notEmpty条件的线程。

offer函数：

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();
            }
        }

说明：offer函数也用于存放元素，在调用ArrayBlockingQueue的add方法时，会间接的调用到offer函数，offer函数添加元素不会抛出异常，当底层Object数组已满时，则返回false，否则，会调用enqueue函数，将元素存入底层Object数组。并唤醒等待notEmpty条件的线程。

take函数

public E take() throws InterruptedException {
            // 可重入锁
            final ReentrantLock lock = this.lock;
            // 如果当前线程未被中断，则获取锁，中断会抛出异常
            lock.lockInterruptibly();
            try {
                while (count == 0) // 元素数量为0，即Object数组为空
                    // 则等待notEmpty条件
                    notEmpty.await();
                // 出队列
                return dequeue();
            } finally {
                // 释放锁
                lock.unlock();
            }
        }

说明：take函数用于从ArrayBlockingQueue中获取一个元素，其与put函数相对应，在当前线程被中断时会抛出异常，并且当队列为空时，会阻塞一直等待。其中，take会调用dequeue函数，dequeue函数源码如下

private E dequeue() {
            // assert lock.getHoldCount() == 1;
            // assert items[takeIndex] != null;
            final Object[] items = this.items;
            @SuppressWarnings("unchecked")
            // 取元素
            E x = (E) items[takeIndex];
            // 该索引的值赋值为null
            items[takeIndex] = null;
            // 取值索引等于数组长度
            if (++takeIndex == items.length)
                // 重新赋值取值索引
                takeIndex = 0;
            // 元素个数减1
            count--;
            if (itrs != null) 
                itrs.elementDequeued();
            // 唤醒在notFull条件上等待的线程
            notFull.signal();
            return x;
        }

说明：dequeue函数用于将取元素，并且会唤醒等待notFull条件的线程。

poll函数

public E poll() {
            // 重入锁
            final ReentrantLock lock = this.lock;
            // 获取锁
            lock.lock();
            try {
                // 若元素个数为0则返回null，否则，调用dequeue，出队列
                return (count == 0) ? null : dequeue();
            } finally {
                // 释放锁
                lock.unlock();
            }
        }

说明：poll函数用于获取元素，其与offer函数相对应，不会抛出异常，当元素个数为0是，返回null，否则，调用dequeue函数，并唤醒等待notFull条件的线程。并返回。

clear函数

public void clear() {
            // 数组
            final Object[] items = this.items;
            // 可重入锁
            final ReentrantLock lock = this.lock;
            // 获取锁
            lock.lock();
            try {
                // 保存元素个数
                int k = count;
                if (k > 0) { // 元素个数大于0
                    // 存数元素索引
                    final int putIndex = this.putIndex;
                    // 取元素索引
                    int i = takeIndex;
                    do {
                        // 赋值为null
                        items[i] = null;
                        if (++i == items.length) // 重新赋值i
                            i = 0;
                    } while (i != putIndex);
                    // 重新赋值取元素索引
                    takeIndex = putIndex;
                    // 元素个数为0
                    count = 0;
                    if (itrs != null)
                        itrs.queueIsEmpty();
                    for (; k > 0 && lock.hasWaiters(notFull); k--) // 若有等待notFull条件的线程，则逐一唤醒
                        notFull.signal();
                }
            } finally {
                // 释放锁
                lock.unlock();
            }
        }

说明：clear函数用于清空ArrayBlockingQueue，并且会释放所有等待notFull条件的线程（存放元素的线程）。

示例1：

package com.test.collections;
 
 import java.util.concurrent.ArrayBlockingQueue;
 
 class PutThread extends Thread {
 private ArrayBlockingQueue<Integer> abq;
 public PutThread(ArrayBlockingQueue<Integer> abq) {
 this.abq = abq;
 }
 
 public void run() {
 for (int i = 0; i < 10; i++) {
 try {
 System.out.println("put " + i);
 abq.put(i);
 Thread.sleep(100);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 }
 }
 }
 
 class GetThread extends Thread {
 private ArrayBlockingQueue<Integer> abq;
 public GetThread(ArrayBlockingQueue<Integer> abq) {
 this.abq = abq;
 }
 
     public void run() {
     	while (true) {
     try {
     // 消费消息
 	System.out.println("take " + abq.take());
 	} catch (InterruptedException e) {
 	// TODO Auto-generated catch block
 	e.printStackTrace();
 	}
 	}
 //        for (int i = 0; i < 10; i++) {
 //            try {
 //                System.out.println("take " + abq.take());
 //                Thread.sleep(100);
 //            } catch (InterruptedException e) {
 //                e.printStackTrace();
 //            }
 //        }
     }
 }
 
 public class ArrayBlockingQueueDemo {
 public static void main(String[] args) {
 ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<Integer>(10);
 PutThread p1 = new PutThread(abq);
 GetThread g1 = new GetThread(abq);
 
 p1.start();
 g1.start();
 }
 }

运行结果：

put 0
    take 0
    put 1
    take 1
    put 2
    take 2
    put 3
    take 3
    put 4
    take 4
    put 5
    take 5
    put 6
    take 6
    put 7
    take 7
    put 8
    take 8
    put 9
    take 9

说明：示例中使用了两个线程，一个用于存元素，一个用于读元素，存和读各10次，每个线程存一个元素或者读一个元素后都会休眠100ms，可以看到结果是交替打印，并且首先打印的肯定是put线程语句（因为若取线程先取元素，此时队列并没有元素，其会阻塞，等待存线程存入元素），并且最终程序可以正常结束。

① 若修改取元素线程，将存的元素的次数修改为15次（for循环的结束条件改为15即可），运行结果如下：

说明：运行结果与上面的运行结果相同，但是，此时程序无法正常结束，因为take方法被阻塞了，等待被唤醒。

示例2：

package com.git.test;
 
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 
 public class WorkerThread {
 
 	public static void main(String[] args) {
 		Helper helper = new Helper();
 		helper.init();
 
 		// 此处，helper的客户端线程为main线程
 		helper.submit("Something...");
 
 	}
 
 	static class Helper {
 		private final BlockingQueue<String> workQueue = new ArrayBlockingQueue<String>(
 		 100);
 
 		// 用于处理队列workQueue中的任务的工作者线程
 		private final Thread workerThread = new Thread() {
 
 			@Override
 			public void run() {
 				String task = null;
 				while (true) {
 					try {
 						task = workQueue.take();
 					} catch (InterruptedException e) {
 						break;
 					}
 					System.out.println(doProcess(task));
 
 				}
 			}
 
 		};
 
 		public void init() {
 			workerThread.start();
 		}
 
 		protected String doProcess(String task) {
 			return task + "->processed.";
 		}
 
 		public void submit(String task) {
 			try {
 				workQueue.put(task);
 			} catch (InterruptedException e) {
 				Thread.currentThread().interrupt();
 				;
 			}
 		}
 
 	}
 
 }

总结：

总的来说，有了前面分析的基础，分析ArrayBlockingQueue就会非常的简单，ArrayBlockingQueue是通过ReentrantLock和Condition条件来保证多线程的正确访问的。ArrayBockingQueue的分析就到这里

参考：

http://www.cnblogs.com/leesf456/p/5533770.html jdk1.8版本

http://blog.csdn.net/xin\_jmail/article/details/26157971 jdk1.7版本