Android并发学习之阻塞队列

简介

多线程环境中，经过队列能够很容易实现数据共享，好比经典的“生产者”和“消费者”模型中，经过队列能够很便利地实现二者之间的数据共享。假设咱们有若干生产者线程，另外又有若干个消费者线程。若是生产者线程须要把准备好的数据共享给消费者线程，利用队列的方式来传递数据，就能够很方便地解决他们之间的数据共享问题。但若是生产者和消费者在某个时间段内，万一发生数据处理速度不匹配的状况呢？理想状况下，若是生产者产出数据的速度大于消费者消费的速度，而且当生产出来的数据累积到必定程度的时候，那么生产者必须暂停等待一下（阻塞生产者线程），以便等待消费者线程把累积的数据处理完毕，反之亦然。

BlockingQueue很好地解决了上述问题，BlockingQueue即阻塞队列，它是一个接口，它的实现类有ArrayBlockingQueue、DelayQueue、 LinkedBlockingDeque、LinkedBlockingQueue、PriorityBlockingQueue、SynchronousQueue等，它们的区别主要体如今存储结构上或对元素操做上的不一样。

经常使用方法

public interface BlockingQueue<E> extends Queue<E> {

    //往队列尾部添加元素，若是BlockingQueue能够容纳,则返回true,不然抛出异常
    boolean add(E e);
    
   //移除元素，若是有这个元素则就回true,不然抛出异常 
    boolean remove(Object o);
     
    //往队列尾部添加元素,若是BlockingQueue能够容纳则返回true,不然返回false.
    //若是是往限定了长度的队列中设置值，推荐使用offer()方法。
    boolean offer(E e);
    
   //和上面的方法差很少，不过若是队列满了能够阻塞等待一段时间
    boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException;
    
    //取出头部对象,若不能当即取出,则能够等time参数规定的时间,取不到时返回null
    E poll(long timeout, TimeUnit unit) throws InterruptedException; 
    
     //往队列尾部添加元素,若是没有空间,则调用此方法的线程被阻塞直到有空间再继续. 
    void put(E e) throws InterruptedException;
    
    //取出头部对象,若BlockingQueue为空,阻断进入等待状态直到Blocking有新的对象被加入为止 
    E take() throws InterruptedException;
    
    //剩余容量，超出此容量，便没法无阻塞地添加元素
    int remainingCapacity();
    
    //判断队列中是否拥有该值。
    boolean contains(Object o);
    
    //一次性从BlockingQueue获取全部可用的数据对象,能够提高获取数据效率
    int drainTo(Collection<? super E> c);
    
     //和上面的方法差很少，不过限制了最大取出数量
    int drainTo(Collection<? super E> c, int maxElements);
    
}
复制代码

源码简析

咱们以ArrayBlockingQueue为例分析下上述方法:

offer(E e)

public boolean offer(E e) {
        Objects.requireNonNull(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == items.length)
                return false;
            else {
                enqueue(e);
                return true;
            }
        } finally {
            lock.unlock();
        }
    }
    
   private void enqueue(E x) {
        final Object[] items = this.items;
        items[putIndex] = x;
        if (++putIndex == items.length) putIndex = 0;
        count++;
        notEmpty.signal();
    }
复制代码

offer操做如上，代码比较简单，可见阻塞队列是经过可重入保证线程安全。enqueue方法也说明了ArrayBlockingQueue是经过数组的形式存储数据的。若是队列满了直接会返回false，不会阻塞线程。

put(E e)

public void put(E e) throws InterruptedException {
        Objects.requireNonNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == items.length)//队列满了，一直阻塞在这里
                notFull.await();
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }

复制代码

由于put方法在队列已满的状况下会阻塞线程，take、poll等方法会调用dequeue方法出列，从而调用notFull.signal()，从而唤醒阻塞在put方法中线程去继续进行入列操做：

take()

public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }
  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--;
        if (itrs != null)
            itrs.elementDequeued();
        notFull.signal();
        return x;
    }
复制代码

poll(long timeout, TimeUnit unit)

从对头取出一个元素：若是数组不空，出队；若是数组已空且已经超时，返回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 <= 0L)
                    return null;
                //阻塞指定时间，enqueue()方法会调用notEmpty.signal()唤醒进行poll操做的线程
                nanos = notEmpty.awaitNanos(nanos);
            }
            return dequeue();
        } finally {
            lock.unlock();
        }
    }
复制代码

参考文章和扩展阅读

虽然只讲了阻塞队列，但涉及了ReentrantLock、中断、Condition等知识点，若是不清楚的话能够看下下面的几篇文章：

blog.csdn.net/vernonzheng…

wsmajunfeng.iteye.com/blog/162935…

Lock锁和Condition条件

ava中Lock，tryLock，lockInterruptibly有什么区别？