Java阻塞队列

Java中的阻塞队列

1.什么是阻塞队列

阻塞队列（BlockingQueue)，就是支持阻塞插入和阻塞移除的队列。
1）阻塞插入：当队列满时，会阻塞插入元素的线程，知道队列不为满时。
2) 阻塞移除：当队列为空时，获取元素的线程会阻塞，等待队列变为非空。

2.运用场景

阻塞队列可用于生产者和消费者问题，生产者就是向队列里插入元素，而消费者就是从队列里获取元素。这时，阻塞队列就可以充当生产者存元素、消费者取元素的容器。

3.消息介绍

方法/处理方式	抛出异常	返回特殊值	一直阻塞	超时退出
插入方法	add(e)	offer(e)	put(e)	offer(e,time,unit)
移除方法	remove()	poll()	take()	poll(time,unit)
检查方法	element()	peek()	不可用	不可用

抛出异常: 当队列满插入或者队列空移除时，会抛出相应的异常。
返回特殊值：插入队列返回插入是否成功，移除队列，返回相应的值，没有则返回null。
一直阻塞：也就上面的消费者和生产者的情况，队列满插入则阻塞，队列空移除也是阻塞。
超时退出：同样的道理，也是会阻塞一段时间，超时则线程退出。

4.JDK提供的阻塞队列API

jdk1.8中的阻塞队列，接口BlockingQueue,继承结构

ArrayBlockingQueue: 一个由数组结构组成的有界阻塞队列
LinkedBlockingQueue: 链表结构的有界阻塞队列
PriorityBlockingQueue: 支持优先级排序的无界阻塞队列
DelayQueue: 使用优先级队列实现的无界阻塞队列
SynchronousQueue: 不存储元素的阻塞队列
LinkedTransferQueue: 链表结构的无界阻塞队列
LinkedBlockingDeque: 链表结构的双向阻塞队列

5.讨论ArrayBlockingQueue

阻塞队列很多，我们拿一个来举例；
JDK8中的部分源码
三个构造方法，可以发现，ArrayBlockingQueue必须知道大小，默认是非公平的

 /** * Creates an {@code ArrayBlockingQueue} with the given (fixed) * capacity and default access policy. * * @param capacity the capacity of this queue * @throws IllegalArgumentException if {@code capacity < 1} */
    public ArrayBlockingQueue(int capacity) {
   
        this(capacity, false);
    }

    /** * 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();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

    /** * Creates an {@code ArrayBlockingQueue} with the given (fixed) * capacity, the specified access policy and initially containing the * elements of the given collection, * added in traversal order of the collection's iterator. * * @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. * @param c the collection of elements to initially contain * @throws IllegalArgumentException if {@code capacity} is less than * {@code c.size()}, or less than 1. * @throws NullPointerException if the specified collection or any * of its elements are null */
    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);
                    items[i++] = e;
                }
            } catch (ArrayIndexOutOfBoundsException ex) {
   
                throw new IllegalArgumentException();
            }
            count = i;
            putIndex = (i == capacity) ? 0 : i;
        } finally {
   
            lock.unlock();
        }
    }

一些属性：可以发现，使用了ReentrantLock来实现访问公平性。

 /** 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;

提问：如果队列是空的，消费者一直等待，当生产者添加元素时，消费者如何知道队列由元素？
使用通知模式实现 ，所谓通知模式，就是当生产者往满的队列添加元素时，会发生阻塞，当消费者消费了一个元素后，会通知生产者当前队列可用，JDK源码中使用了Condition来实现：

   /** Condition for waiting takes */
    private final Condition notEmpty;

    /** Condition for waiting puts */
    private final Condition notFull;
    // 省略 代码
 	public ArrayBlockingQueue(int capacity, boolean fair) {
   
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }
    // 省略代码
      /** * Inserts the specified element at the tail of this queue, waiting * for space to become available if the queue is full. * * @throws InterruptedException {@inheritDoc} * @throws NullPointerException {@inheritDoc} */
    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();
        }
    }
    // 省略代码
    public E take() throws InterruptedException {
   
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
   
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
   
            lock.unlock();
        }
    }

6.阻塞队列演示消费者与生产者模型

public class TestConsumerAndProducerBlockingQueue {
   

    public static void main(String[] args) {
   
        // 进行测试 创建一个销售员对象
        ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);

        // 创建两个消费者，两个生产者
        new Thread(new TestConsumer(queue),"消费者A").start();
        new Thread(new TestProducer(queue),"生产者B").start();
        new Thread(new TestConsumer(queue),"消费者C").start();
        new Thread(new TestProducer(queue),"生产者D").start();

    }

}

// 定义消费者
class TestConsumer implements Runnable {
   

    private ArrayBlockingQueue<Integer> product;

    public TestConsumer(ArrayBlockingQueue queue) {
   
        this.product = queue;
    }

    @Override
    public void run() {
   
        // 进行十次消费
        for (int i = 0; i < 10; i++) {
   
            try {
   
                System.out.println(Thread.currentThread().getName() + "消费了：" + product.take());
            } catch (InterruptedException e) {
   
                e.printStackTrace();
            }
        }
    }
}

// 定义生产者
class TestProducer implements Runnable {
   

    private ArrayBlockingQueue<Integer> product;

    public TestProducer(ArrayBlockingQueue queue) {
   
        this.product = queue;
    }

    @Override
    public void run() {
   
        // 进行十次生产
        for (int i = 0; i < 10; i++) {
   
            try {
   
                System.out.println(Thread.currentThread().getName() + "生成产品：" + i);
                product.put(i);
            } catch (InterruptedException e) {
   
                e.printStackTrace();
            }
        }
    }
}