juc-10-阻塞队列

也许你以前学习数据结构时，接触过Queue(队列)，它具有先进先出的特性。

什么是阻塞队列 BlockingQueue ？
顾名思义它一个队列，具有阻塞功能。

为什么要学习阻塞队列？

1. 很多朋友可能有些陌生，其实只要使用过线程池，就表示你已经间接地使用过阻塞队列了，阻塞队列是线程池的重要组成部分，后面文章会详细介绍线程池。
2. 阻塞队列是线程安全的，所以阻塞队列可以用作线程安全的并发容器，比如生产者消费者模式。

阻塞队列关系图

由数组结构组成的有界阻塞队，需要指定队列容量，可指定公平策略：如果是公平的策略，那么等待了最长时间的线程会优先被处理，不过这会同时带来一定的性能损耗

由链表结构组成的有界阻塞队列，如果不指定容量，默认为Integer.MAX_VALUE。
内部结构：Node
两把锁Lock：takeLock、putLock
两个条件Condition：notEmpty、notFull

使用优先级队列实现的无界阻塞队列，支持优先级，自然顺序（而不是先进先出），是 priorityQueue的线程安全版本

一个不存储元素的阻塞队列，它的容量为 0，每一个put操作都要等待一个take操作。

注意：

• SynchronousQueue 的容量不是1而是0，因为 SynchronousQueue 不需要去持有元素，它所做的就是直接传递（direct handoff）
• SynchronousQueue 没有 peek等函数，因为 peek 的含义是取出头节点，但是 SynchronousQueue 的容量为 0，所以连头节点都没有，也就没有 peek 方法。同理，没有 iterate 相关方法。
• 是一个极好的用来直接传递的并发数据结构
• SynchronousQueue 是线程池 Executors.newCachedThreadPool() 使用的阻塞队列。

使用优先级队列实现的无界阻塞队列，支持延时获取的元素的阻塞队列，元素必须要实现Delayed接口，获取元素剩余的延迟时间
适用场景：订单到期，限时支付等等

下面以生产者消费者案例为场景，演示 ArrayBlockingQueue 阻塞队列的使用。

生产者就是生产数据的线程，消费者就是消费数据的线程。
在多线程开发中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这种生产消费能力不均衡的问题，便有了生产者和消费者模式。
生产者和消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。

生产者和消费者彼此之间不直接通信，而是通过阻塞队列来进行通信，生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

ArrayBlockingQueue实现生产者和消费者模式

public class ArrayBlockingQueueDemo {

    //用阻塞队列作为容器，指定容量为5
    private ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(5);

    // 全局商品索引
    static int goodsIndex;

    // 打印日志的lock，是公平锁，等待时间长的线程优先拿到锁
    private Lock fairLock = new ReentrantLock(true);

    public static void main(String[] args) {
 ArrayBlockingQueueDemo demo = new ArrayBlockingQueueDemo();
 for (int i = 0; i < 3; i++) {
     Producer producer = new Producer(demo);
     new Thread(producer, "生产者P" + (i + 1)).start();
 }
 for (int i = 0; i < 2; i++) {
     Consumer consumer = new Consumer(demo);
     new Thread(consumer, "消费者C" + (i + 1)).start();
 }
    }

    
    class Goods {
 // 商品下标
 String index;
    }

    
    public void produce() throws InterruptedException {
 Goods goods = new Goods();
 // 如果有空间，队尾插入数据；如果队列已满，则无法插入，阻塞，直到有空闲时间
 this.blockingQueue.put(goods);
 this.log(goods, true);//打印生产日志
    }

    
    public void consume() throws InterruptedException {
 // 获取并删除队头节点，
如果队列无数据，则阻塞，直到有数据
 Goods goods = this.blockingQueue.take();
 this.log(goods, false);//打印消费日志
    }

    
    private void log(Goods goods, boolean isProduct) {
 this.fairLock.lock();
 try {
     String flag = isProduct ? "生产--" : "消费--";
     if (isProduct) {
  //如果是生产日志，这里设置商品下标
  goods.index = "商品" + goodsIndex++;
     }
     System.out.println(Thread.currentThread().getName() + flag + goods.index);
 } finally {
     this.fairLock.unlock();
 }
    }

}

//生产者
class Producer implements Runnable {
    private ArrayBlockingQueueDemo demo;

    public Producer(ArrayBlockingQueueDemo demo) {
 this.demo = demo;
    }

    @Override
    public void run() {
 while (true) {

     try {
  // 生产者生产
  this.demo.produce();
  TimeUnit.MILLISECONDS.sleep(3000);//模拟生产耗时
     } catch (InterruptedException e) {
  e.printStackTrace();
     }
 }
    }
}

//消费者
class Consumer implements Runnable {
    private ArrayBlockingQueueDemo demo;

    public Consumer(ArrayBlockingQueueDemo demo) {
 this.demo = demo;
    }

    @Override
    public void run() {
 while (true) {

     try {
  // 消费者消费
  this.demo.consume();
  TimeUnit.MILLISECONDS.sleep(2000);//模拟消费耗时
     } catch (InterruptedException e) {
  e.printStackTrace();
     }
 }
    }
}

生产者P1生产--商品0
生产者P2生产--商品1
消费者C2消费--商品1
消费者C1消费--商品0
生产者P3生产--商品2
消费者C1消费--商品2
生产者P3生产--商品3
生产者P2生产--商品4
消费者C2消费--商品3
生产者P1生产--商品5
消费者C1消费--商品4
消费者C2消费--商品5
生产者P1生产--商品6
生产者P3生产--商品7
消费者C1消费--商品6
生产者P2生产--商品8
消费者C2消费--商品7
消费者C1消费--商品8
生产者P2生产--商品9
生产者P1生产--商品10
...

    //使用数组来存储队列元素
    final Object[] items;
    
    //下一次take, poll, peek or remove操作对应在 items数组下标，也就是队头下标
    int takeIndex;

    //下一次 put, offer, or add 操作对应在 items数组下标，也就是队尾下标
    int putIndex;

    //队列中元素的数量
    int count;

    //用于控制所有访问操作的锁
    final ReentrantLock lock;

    // 队列为空时，控制 take 操作等待的条件
    private final Condition notEmpty;

    // 队列满时，控制 put 操作等待的条件
    private final Condition notFull;

    
    public ArrayBlockingQueue(int capacity) {
 this(capacity, false);
    }

    
    public ArrayBlockingQueue(int capacity, boolean fair) {
 if (capacity <= 0)
     throw new IllegalArgumentException();
 // 初始化数组
 this.items = new Object[capacity];
 // 初始化 Lock 和 Condition
 lock = new ReentrantLock(fair);
 notEmpty = lock.newCondition();
 notFull =  lock.newCondition();
    }

    
    public ArrayBlockingQueue(int capacity, boolean fair,
  Collection 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;
     // 如果 Collecion的元素个数超过 items 的容量，则后面加入的元素会覆盖最开始的元素
     putIndex = (i == capacity) ? 0 : i;
 } finally {
     lock.unlock();
 }
    }

    
    public void put(E e) throws InterruptedException {
 //入队元素不能为空
 checkNotNull(e);
 final ReentrantLock lock = this.lock;
 //获取锁
 lock.lockInterruptibly();
 try {
     // 如果队列已满，notFull.await();释放锁，阻塞当前线程等待
     while (count == items.length)
  notFull.await();
     // 队列不满，执行入队操作    
     enqueue(e);
 } finally {
     // 释放锁
     lock.unlock();
 }
    }

    //在 putIndex 下标位置，插入元素x，putIndex就是队尾下标
    //并且执行 notEmpty.signal(); 唤醒因为队列为空执行take操作阻塞等待的线程
    //需要获取锁后，才能调用这个方法
    private void enqueue(E x) {
 // assert lock.getHoldCount() == 1;
 // assert items[putIndex] == null;
 final Object[] items = this.items;
 //在 putIndex 下标位置，插入元素x
 items[putIndex] = x;
 if (++putIndex == items.length)
     putIndex = 0;
 count++;//当前队列元素+1
 // notEmpty Condition signal，唤醒因为 notEmpty.await() 的线程，也就是队列为空执行take操作阻塞等待的线程
 notEmpty.signal();
    }

    //在 putIndex 下标位置，插入元素x，
    //并且执行 notEmpty.signal(); 唤醒因为队列为空执行take操作阻塞等待的线程
    //需要获取锁后，才能调用这个方法
    private void enqueue(E x) {
 // assert lock.getHoldCount() == 1;
 // assert items[putIndex] == null;
 final Object[] items = this.items;
 //在 putIndex 下标位置，插入元素x
 items[putIndex] = x;
 if (++putIndex == items.length)
     putIndex = 0;
 count++;//当前队列元素+1
 // notEmpty Condition signal，唤醒因为 notEmpty.await() 的线程，也就是队列为空执行take操作阻塞等待的线程
 notEmpty.signal();
    }
    
    
    public boolean offer(E e) {
 //入队元素不能为空
 checkNotNull(e);
 final ReentrantLock lock = this.lock;
 //获取锁
 lock.lock();
 try {
     // 队列已满，返回false
     if (count == items.length)
  return false;
     else {
  // 队列不满，执行入队操作，返回true
  enqueue(e);
  return true;
     }
 } finally {
     //释放锁
     lock.unlock();
 }
    }

    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) {
  //等待时间到，返回false
  if (nanos <= 0)
      return false;
  //notFull.await();释放锁，并且指定当前线程阻塞等待时长 nanos      
  nanos = notFull.awaitNanos(nanos);
     }
     // 队列不满，执行入队操作，返回true
     enqueue(e);
     return true;
 } finally {
     //释放锁
     lock.unlock();
 }
    }

    //如果有空间，队尾插入数据，返回true；如果队列已满，则抛出 IllegalStateException
    public boolean add(E e) {
 // 调用 offer(e)，如果有空间，队尾插入数据，返回true
 if (offer(e))
     return true;
 else // 如果队列已满 offer(e) 返回false，抛出 IllegalStateException 
     throw new IllegalStateException("Queue full");
    }

    // 返回并删除 items 数组的takeIndex下标的元素，takeIndex就是队头下标
    // 需要持有锁，才能调用该方法
    private E dequeue() {
 // assert lock.getHoldCount() == 1;
 // assert items[takeIndex] != null;
 final Object[] items = this.items;
 @SuppressWarnings("unchecked")
 // 获取takeIndex下标的元素
 E x = (E) items[takeIndex];
 // 将该下标的元素设为null，表示删除队头
 items[takeIndex] = null;
 if (++takeIndex == items.length)
     takeIndex = 0;
 // 队列元素减一    
 count--;
 if (itrs != null)
     itrs.elementDequeued();
 // notFull Condition signal，唤醒因为 notFull.await() 的线程，也就是队列满了，执行put操作阻塞等待的线程    
 notFull.signal();
 return x;
    }

    // 获取并删除队头节点，如果队列无数据，则阻塞，直到有数据  
    public E take() throws InterruptedException {
 final ReentrantLock lock = this.lock;
 lock.lockInterruptibly();
 //获取锁
 try {
     // 队列为空，notEmpty.await();释放锁，阻塞当前线程等待
     while (count == 0)
  notEmpty.await();
     //队列不为空，获取并删除队头节点，并唤醒因为 notFull.await() 的线程，也就是队列满了，执行put操作阻塞等待的线程     
     return dequeue();
 } finally {
     lock.unlock();
 }
    }

    // 返回并删除 items 数组的takeIndex下标的元素，takeIndex就是队头下标
    // 需要持有锁，才能调用该方法
    private E dequeue() {
 // assert lock.getHoldCount() == 1;
 // assert items[takeIndex] != null;
 final Object[] items = this.items;
 @SuppressWarnings("unchecked")
 // 获取takeIndex下标的元素
 E x = (E) items[takeIndex];
 // 将该下标的元素设为null，表示删除队头
 items[takeIndex] = null;
 if (++takeIndex == items.length)
     takeIndex = 0;
 // 队列元素减一    
 count--;
 if (itrs != null)
     itrs.elementDequeued();
 // notFull Condition signal，唤醒因为 notFull.await() 的线程，也就是队列满了，执行put操作阻塞等待的线程    
 notFull.signal();
 return x;
    }

    // 获取并删除队头节点，如果队列无数据，则返回 null
    public E poll() {
 final ReentrantLock lock = this.lock;
 // 获取锁
 lock.lock();
 try {
     // 如果队列为空，返回null
     //队列不为空，获取并删除队头节点，并唤醒因为 notFull.await() 的线程，也就是队列满了，执行put操作阻塞等待的线程 
     return (count == 0) ? null : 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) {
  //等待时间到，队列依然为空，返回null
  if (nanos <= 0)
      return null;
  //notEmpty.await();释放锁，并且指定当前线程阻塞等待时长 nanos      
  nanos = notEmpty.awaitNanos(nanos);
     }
     //队列不为空，获取并删除队头节点，并唤醒因为 notFull.await() 的线程，也就是队列满了，执行put操作阻塞等待的线程 
     return dequeue();
 } finally {
     lock.unlock();
 }
    }

    //获取并删除队头节点，如果队列无数据，则抛出 NoSuchElementException 
    public E remove() {
 // 调用poll()方法：获取并删除队头节点，如果队列无数据，则返回 null
 E x = poll();
 if (x != null) // 队列不为空，返回队头
     return x;
 else // 如果队列无数据，poll()方法返回null，则抛出 NoSuchElementException
     throw new NoSuchElementException();
    }

    //获取但是不删除队头节点，如果队列无数据，返回null 
    public E peek() {
 final ReentrantLock lock = this.lock;
 //获取锁
 lock.lock();
 try {
     // 返回items中下标takeIndex的元素，如果队列无数据，返回null 
     return itemAt(takeIndex); 
 } finally {
     lock.unlock();
 }
    }

    
    @SuppressWarnings("unchecked")
    final E itemAt(int i) {
 return (E) items[i];
    }

    //获取但是不删除队头节点，如果队列无数据，抛出 NoSuchElementException  
    public E element() {
 // 调用 peek() 方法，获取但不删除 items 数组中下标takeIndex的元素，也就是队尾元素，如果队列无数据，返回null 
 E x = peek();
 if (x != null)
     return x;// x!=null,则返回 x
 else // x==null 抛出 NoSuchElementException  
     throw new NoSuchElementException();
    }