java中BlockingQueue如何实现生产者消费者

这篇文章主要为大家展示了“java中BlockingQueue如何实现生产者消费者”，内容简而易懂，条理清晰，希望能够帮助大家解决疑惑，下面让小编带领大家一起研究并学习一下“java中BlockingQueue如何实现生产者消费者”这篇文章吧。

1.介绍

     阻塞队列 (BlockingQueue)是Java util.concurrent包下重要的数据结构，BlockingQueue提供了线程安全的队列访问方式：当阻塞队列进行插入数据时，如果队列已满，线程将会阻塞等待直到队列非满；从阻塞队列取数据时，如果队列已空，线程将会阻塞等待直到队列非空。并发包下很多高级同步类的实现都是基于BlockingQueue实现的。

jdk7提供了以下7个阻塞队列：

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

阻塞队列提供了下列四种处理方法:

这4类方法中，在队列已满（或为空）的情况下，有些会抛出异常，有些则返回true/false，有些则一直阻塞，还有些则可以设置超时时间，时间到了后，自动退出阻塞状态，实际项目中可根据需要选取适合的方法。

2.队列

队列(Queue)与栈(Stack)是数据结构中的二种常用结构，队列的特点是先进先出(FirstInFirstOut)，而Stack是先进后出(FirstInLastOut)，说得通俗点：Queue就是电影院入场时人们排起来的进场队伍，先来的人（即：前排在前面的人）先入场，而Statck则是一队人依次进入了一个死胡同想出来，先进去(最里面)的人，必须等后面的人(后进入的人)出来了，自己才能出来。

阻塞队列与普通队列的区别在于，当队列是空的时，从队列中获取元素的操作将会被阻塞，或者当队列是满时，往队列里添加元素的操作会被阻塞。试图从空的阻塞队列中获取元素的线程将会被阻塞，直到其他的线程往空的队列插入新的元素。同样，试图往已满的阻塞队列中添加新元素的线程同样也会被阻塞，直到其他的线程使队列重新变得空闲起来，如从队列中移除一个或者多个元素，或者完全清空队列.

队列在多线程应用中，常用于生产-消费场景，打个通俗的比方：很多人早上喜欢去买油条，买油条的人相当于消费者，做油条的师傅则是生产者。而油锅边上用于放油条的铁架子，可以看成一个共享的队列，师傅做好油条后，一根一根的捞出来放在架子上，而顾客则按排队的顺序一根根的付好钱从架子上拿。即：队列的一头，不断有人在放入东西（生产元素），另一头不断有人的消费(拿走元素)。这里就有一个很有趣的现象，如果买的人多，师傅来不及做，那么第一个顾客就会一直等着（后面的所有人也得等着，或称为阻塞了后面的人），直到师傅炸好一根，然后第一个顾客买完走了，后面的人才能顶上来，类似的道理，如果架子放满了，没有人来买，师傅就会停下来，等有人来买了，才会继续做，这就是所谓的队列阻塞，而能产生阻塞行为的队列称为阻塞队列。

从刚才的描述可以看出，发生阻塞起码得满足下面至少一个条件：(前提：队列是有界的)

从队列里取元素时，如果队列为空，则代码一直等在这里（即阻塞），直到队列里有东西了，拿到元素了，后面的代码才能继续；

向队列里放元素时，如果队列满了（即放不下更多元素），则代码也会卡住，直到队列里的东西被取走了（即：有空位可以放新元素了），后面的代码才能继续；

3.模拟生产者消费者

模拟了买油条的场景，1个老板在做油条，3个顾客在排队买：

package com.tl.skyLine.thread;import java.util.concurrent.ArrayBlockingQueue;public class BlockingQueueTest {  //final成员变量表示常量，只能被赋值一次，赋值后值不再改变。  private static final int queueSize = 5;  private static final ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<String>(queueSize);  private static final int produceSpeed = 2000;//生产速度(越小越快)  private static final int consumeSpeed = 10;//消费速度(越小越快)  //生产者  public static class Producer implements Runnable {    @Override    public void run() {      while (true) {        try {          System.out.println("老板准备炸油条了，架子上还能放：" + (queueSize - queue.size()) + "根油条");          queue.put("1根油条");          System.out.println("老板炸好了1根油条，架子上还能放：" + (queueSize - queue.size()) + "根油条");          Thread.sleep(produceSpeed);        } catch (InterruptedException e) {          e.printStackTrace();        }      }    }  }  //消费者  public static class Consumer implements Runnable {    @Override    public void run() {      while (true) {        try {          System.out.println("A 准备买油条了，架子上还剩" + queue.size() + "根油条");          queue.take();          System.out.println("A 买到1根油条，架子上还剩" + queue.size() + "根油条");          Thread.sleep(consumeSpeed);          System.out.println("B 准备买油条了，架子上还剩" + queue.size() + "根油条");          queue.take();          System.out.println("B 买到1根油条，架子上还剩" + queue.size() + "根油条");          Thread.sleep(consumeSpeed);          System.out.println("C 准备买油条了，架子上还剩" + queue.size() + "根油条");          queue.take();          System.out.println("C 买到1根油条，架子上还剩" + queue.size() + "根油条");          Thread.sleep(consumeSpeed);        } catch (InterruptedException e) {          e.printStackTrace();        }      }    }  }  public static void main(String[] args) {    Thread producer = new Thread(new Producer());    Thread consumer = new Thread(new Consumer());    producer.start();    consumer.start();  }}

结果：

老板准备炸油条了，架子上还能放：5根油条老板炸好了1根油条，架子上还能放：4根油条A 准备买油条了，架子上还剩1根油条A 买到1根油条，架子上还剩0根油条B 准备买油条了，架子上还剩0根油条老板准备炸油条了，架子上还能放：5根油条老板炸好了1根油条，架子上还能放：4根油条B 买到1根油条，架子上还剩0根油条C 准备买油条了，架子上还剩0根油条老板准备炸油条了，架子上还能放：5根油条老板炸好了1根油条，架子上还能放：4根油条C 买到1根油条，架子上还剩0根油条A 准备买油条了，架子上还剩0根油条老板准备炸油条了，架子上还能放：5根油条老板炸好了1根油条，架子上还能放：4根油条A 买到1根油条，架子上还剩0根油条B 准备买油条了，架子上还剩0根油条老板准备炸油条了，架子上还能放：5根油条老板炸好了1根油条，架子上还能放：4根油条

上面设置消费速度快于生产速度，所有会发现控制台中，老板炸油条时，架子上油条卖得快，所以能放4根或者5根油条，而顾客由于购买速度快，所以会经常出现架子上还剩0根油条，当我们把生产速度produceSpeed跟消费速度consumeSpeed对换一下，又会发现相反的结果。

4.原理分析

我们简单模拟一下BlockingQueue，熟悉一下背后的原理：

public class BlockingQueue {private List queue = new LinkedList();private int limit = 10;public BlockingQueue(int limit){this.limit = limit;}public synchronized void enqueue(Object item)throws InterruptedException {while(this.queue.size() == this.limit) {wait();}if(this.queue.size() == 0) {notifyAll();}this.queue.add(item);}public synchronized Object dequeue()throws InterruptedException{while(this.queue.size() == 0){wait();}if(this.queue.size() == this.limit){notifyAll();}return this.queue.remove(0);}}

我们看一下jdk的实现过程：

ArrayBlockingQueue的部分源码：

 final ReentrantLock lock;  private final Condition notEmpty;  private final Condition notFull;

这3个变量很重要，ReentrantLock重入锁，notEmpty检查不为空的Condition以及notFull用来检查队列未满的Condition

Condition是一个接口，里面有二个重要的方法：

await()：Causesthecurrentthreadtowaituntilitissignalledorinterrupted.即阻塞当前线程，直到被通知（唤醒）或中断

singal():Wakesuponewaitingthread.唤醒阻塞的线程

再来看put方法：(jdk1.8)

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

先获取锁
2.然后用while循环检测元素个数是否等于items长度，如果相等，表示队列满了,调用notFull的await()方法阻塞线程
3.否则调用enqueue()方法添加元素
4.最后解锁

这是添加元素的代码（jdk 1.8）,注意最后一行notEmpty.signal()方法，表示添加完元素后，调用singal()通知等待（从队列中取元素）的线程，队列不空(有值)啦，可以来取东西了。

类似的take()与dequeue()方法则相当于逆过程（注：同样都是jdk 1.8）

public E take() throws InterruptedException {  final ReentrantLock lock = this.lock;  lock.lockInterruptibly();  try {    while (count == 0)      notEmpty.await();    return dequeue();  } finally {    lock.unlock();  }}

类似的：

先加锁
2. 如果元素个数为空，表示队列已空，调用notEmpty的await()阻塞线程，直接队列里又有新元素加入为止
3. 然后调用dequeue 从队列里删除元素
4. 解锁

dequeue方法：

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)    itrs.elementDequeued();  notFull.signal();  return x;}

倒数第2行，元素移除后，调用notFull.singnal唤醒等待（向队列添加元素的）线程，队列有空位了，可以向里面添加元素了。

以上是“java中BlockingQueue如何实现生产者消费者”这篇文章的所有内容，感谢各位的阅读！相信大家都有了一定的了解，希望分享的内容对大家有所帮助，如果还想学习更多知识，欢迎关注编程网精选频道！