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则是一队人依次进入了一个死胡同想出来,先进去(最里面)的人,必须等后面的人(后进入的人)出来了,自己才能出来。
阻塞队列与普通队列的区别在于,当队列是空的时,从队列中获取元素的操作将会被阻塞,或者当队列是满时,往队列里添加元素的操作会被阻塞。试图从空的阻塞队列中获取元素的线程将会被阻塞,直到其他的线程往空的队列插入新的元素。同样,试图往已满的阻塞队列中添加新元素的线程同样也会被阻塞,直到其他的线程使队列重新变得空闲起来,如从队列中移除一个或者多个元素,或者完全清空队列.
队列在多线程应用中,常用于生产-消费场景,打个通俗的比方:很多人早上喜欢去买油条,买油条的人相当于消费者,做油条的师傅则是生产者。而油锅边上用于放油条的铁架子,可以看成一个共享的队列,师傅做好油条后,一根一根的捞出来放在架子上,而顾客则按排队的顺序一根根的付好钱从架子上拿。即:队列的一头,不断有人在放入东西(生产元素),另一头不断有人的消费(拿走元素)。这里就有一个很有趣的现象,如果买的人多,师傅来不及做,那么第一个顾客就会一直等着(后面的所有人也得等着,或称为阻塞了后面的人),直到师傅炸好一根,然后第一个顾客买完走了,后面的人才能顶上来,类似的道理,如果架子放满了,没有人来买,师傅就会停下来,等有人来买了,才会继续做,这就是所谓的队列阻塞,而能产生阻塞行为的队列称为阻塞队列。
从刚才的描述可以看出,发生阻塞起码得满足下面至少一个条件:(前提:队列是有界的)
1.从队列里取元素时,如果队列为空,则代码一直等在这里(即阻塞),直到队列里有东西了,拿到元素了,后面的代码才能继续;
2.向队列里放元素时,如果队列满了(即放不下更多元素),则代码也会卡住,直到队列里的东西被取走了(即:有空位可以放新元素了),后面的代码才能继续;
3.模拟生产者消费者
模拟了买油条的场景,1个老板在做油条,3个顾客在排队买:
package com.tl.skyline.thread; import java.util.concurrent.arrayblockingqueue; /** * created by tl on 17/3/3. */ 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的部分源码:
/** main lock guarding all access */ final reentrantlock lock; /** condition for waiting takes */ private final condition notempty; /** condition for waiting puts */ 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(); } }
1.先获取锁
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(); } }
类似的:
1. 先加锁
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实现生产者消费者详解的全部内容,希望对大家有所帮助。感兴趣的朋友可以继续参阅本站:java并发之嵌套管程锁死详解、、java并发编程semaphore计数信号量详解等,有什么问题可以随时留言,小编会及时回复大家的。感谢朋友们对本站的支持!
推荐阅读
-
java并发学习之BlockingQueue实现生产者消费者详解
-
Java实现生产者消费者问题与读者写者问题详解
-
并发编程(九)—— Java 并发队列 BlockingQueue 实现之 LinkedBlockingQueue 源码分析
-
java并发学习之BlockingQueue实现生产者消费者详解
-
JAVA多线程实现生产者消费者的实例详解
-
Java实现生产者消费者问题与读者写者问题详解
-
JAVA多线程实现生产者消费者的实例详解
-
并发编程(十)—— Java 并发队列 BlockingQueue 实现之 SynchronousQueue源码分析
-
Java高并发BlockingQueue重要的实现类详解
-
并发编程(八)—— Java 并发队列 BlockingQueue 实现之 ArrayBlockingQueue 源码分析