玩转并发：深入理解阻塞队列

前言

阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。这两个附加的操作支持阻塞的插入和移除方法。

支持阻塞的插入方法：意思是当队列满时，队列会阻塞插入的元素，直到队列不满支持阻塞的移除方法：意思是在队列为空时，获取元素的线程会等待队列变为非空

阻塞队列常用于生产者和消费者的场景，生产者是向队列里添加元素的线程，消费者从队列里取元素的线程。阻塞队列就是生产者用来存放元素，消费者用来获取元素的容器

插入和移除操作的4种处理方式

方法/处理方式抛出异常返回特殊值一直阻塞超时退出插入方法boolean add©boolean offer(e)void put(e)boolean offer(e, time, unit)解释添加元素，添加成功返回true，如果队列满了，抛出IllegalStateException添加元素，添加成功返回true，如果队列满了，返回false添加元素，如果队列已经满了，则阻塞等待添加元素，添加成功返回true，如果队列已经满了，则阻塞等待，指定时间已经过去还没能添加成功元素，则返回false移除方法E remove()E poll()E take()E poll(time, unit)解释返回头结点，从队列中移除头节点，如果队列为空，抛出NoSuchElementException返回头结点，从队列中移除头节点，如果队列为空，返回null返回头结点，从队列中移除头节点，如果队列为空则阻塞等待返回头结点，从队列中移除头节点，队列中没元素会一直阻塞等待，指定时间已经过去还没能拿到头节点，则返回null检查方法E element()E peek()不可用不可用解释返回头结点，但是不从队列中移除头节点，如果队列为空，抛出NoSuchElementException返回头结点，但是不从队列中移除头节点，如果队列为空，返回null

例子

举一个多生产者，多消费者的例子，队列的大小为3，即队列大小为3时，生产者就不再生产

public class Producer implements Runnable { private ArrayList list; private int capacity; public Producer(ArrayList queue,int capacity) { this.list = queue; this.capacity = capacity; } @Override public void run() { synchronized (list) { while (true) { while (list.size() == capacity) { try { list.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } Object object = list.add(new Object()); System.out.println(Thread.currentThread().getName() + " 生产"); list.notifyAll(); } } } }

注意消费者和生产者都是用的notifyAll()[通知所有阻塞的线程]方法，而不是notify()[通知一个阻塞的线程]方法，因为有可能出现“生产者”唤醒“生产者”，消费者“唤醒”消费者的情况，因此有可能造成死锁，这里以1个消费者，3个生产者为例说一下，消费者1获得锁还没产品呢，阻塞，接着生产者1获得锁生产完了，然后生产者2获得锁后生产完了，再然后生产者3获得锁生产完了，最后生产者1获得锁了，然后阻塞了，现在好了生产者和消费者都阻塞了，造成了死锁。notifyAll()则不会造成死锁，接着上面的步骤，生产者3生产完了释放锁后，会通知所有阻塞的线程，因此消费者1肯定有机会拿到锁来进行消费

public class Consumer implements Runnable { private List list; public Consumer(List queue) { this.list = queue; } @Override public void run() { synchronized (list) { while (true) { while (list.size() == 0) { try { list.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } Object object = list.remove(0); System.out.println(Thread.currentThread().getName() + " 消费"); list.notifyAll(); } } } } public class Test { public static void main(String[] args) { ArrayList list = new ArrayList(3); for (int i = 0; i < 3; i++) { new Thread(new Producer(list, 3), "生产者" + i).start(); new Thread(new Consumer(list), "消费者" + i).start(); } } }

一部分结果

生产者0 生产 生产者0 生产 生产者0 生产 消费者1 消费 消费者1 消费 消费者1 消费 生产者1 生产

把这个实例用阻塞队列来改写，先自己写一个阻塞队列，实现BlockingQueue接口，这里只展示了一部分重写的方法

public class MyBlockingQueue<E> implements BlockingQueue<E> { private int capacity; private List<E> list; public MyBlockingQueue(int capacity) { this.capacity = capacity; this.list = new ArrayList(capacity); } @Override public void put(E e) throws InterruptedException { synchronized (this) { if (list.size() == capacity) { this.wait(); } list.add(e); this.notifyAll(); } } @Override public E take() throws InterruptedException { synchronized (this) { while (list.size() == 0) { this.wait(); } E value = list.remove(0); this.notifyAll(); return value; } } } public class Producer implements Runnable { private BlockingQueue queue; public Producer(BlockingQueue queue) { this.queue = queue; } @Override public void run() { while (true) { try { queue.put(new Object()); System.out.println(Thread.currentThread().getName() + " 生产"); } catch (InterruptedException e) { e.printStackTrace(); } } } } public class Consumer implements Runnable { private BlockingQueue queue; public Consumer(BlockingQueue queue) { this.queue = queue; } @Override public void run() { while (true) { try { Object object = queue.take(); System.out.println(Thread.currentThread().getName() + " 消费"); } catch (InterruptedException e) { e.printStackTrace(); } } } } public class Test { public static void main(String[] args) { BlockingQueue queue = new MyBlockingQueue(3); for (int i = 0; i < 3; i++) { new Thread(new Producer(queue), "生产者" + i).start(); new Thread(new Consumer(queue), "消费者" + i).start(); } } }

这里将BlockingQueue的实现改为ArrayBlockingQueue，程序运行结果一样，和我们之前的例子比较，BlockingQueue其实就是不用我们自己写阻塞和唤醒的部分，直接看一下ArrayBlockingQueue的源码，其实和我自己实现的差不多，只不过是并发这部分源码用的是ReentLock，而我用的是synchronized

源码

基于jdk1.8.0_20 先看属性

public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable { /** 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; /** Main lock guarding all access */ final ReentrantLock lock; /** 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(); }

put方法

public void put(E e) throws InterruptedException { checkNotNull(e); final ReentrantLock lock = this.lock; // 响应中断的方式上锁 lock.lockInterruptibly(); try { while (count == items.length) // 将当前线程加入notFull这个等待队列 notFull.await(); enqueue(e); } finally { lock.unlock(); } } private static void checkNotNull(Object v) { if (v == null) throw new NullPointerException(); } 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) putIndex = 0; count++; notEmpty.signal(); }

take方法

public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == 0) // 将当前线程加入notEmpty这个等待队列 notEmpty.await(); return dequeue(); } finally { lock.unlock(); } } private E dequeue() { // assert lock.getHoldCount() == 1; // assert items[takeIndex] != null; 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) //更新iterators，不再分析 itrs.elementDequeued(); notFull.signal(); return x; }

其他方法就不再分析，基本上就是上锁，然后进行相应的操作 最后说一下LZ的理解，个人感觉用ArrayBlockingQueue实现生产者和消费者，比我上面用synchronized的方式应该快很多，毕竟ArrayBlockingQueue只会是生成者通知消费者，或者消费者通知生产者，而synchronized不是，会造成很多不必要的锁竞争，当然并没有实验，有时间可以试试

参考博客

[1]https://blog.csdn.net/luohuacanyue/article/details/16359777 [2]https://www.cnblogs.com/chentingk/p/6497107.html [3]https://www.cnblogs.com/Ming8006/p/7243858.html 实现阻塞队列 [4]https://blog.csdn.net/truelove12358/article/details/59597169 [5]https://blog.csdn.net/z83986976/article/details/52017259