成都学网站建设费用,电子商务电商网站饿建设,帮助网站源码,建设网站工作室的问题疑问阻塞队列一 阻塞队列1.1.阻塞队列概念1.2.阻塞队列API案例1.2.1. ArrayBlockingQueue1.2.1.1.抛出异常1.2.1.2.返回布尔1.2.1.3.阻塞1.2.1.4.超时1.2.2.SynchronousQueue二 阻塞队列应用---生产者消费者2.1.传统模式案例代码结果案例问题---防止虚假唤醒2.2.⽣产者消费者防⽌虚…
阻塞队列一 阻塞队列1.1.阻塞队列概念1.2.阻塞队列API案例1.2.1. ArrayBlockingQueue1.2.1.1.抛出异常1.2.1.2.返回布尔1.2.1.3.阻塞1.2.1.4.超时1.2.2.SynchronousQueue二 阻塞队列应用---生产者消费者2.1.传统模式案例代码结果案例问题---防止虚假唤醒2.2.⽣产者消费者防⽌虚假唤醒2.2.1 新版⽣产者消费者写法 ReentrantLock.Condition案例代码2.2.2精准通知顺序访问案例代码2.3.Synchronized和Lock的区别2.4. 阻塞队列模式⽣产者消费者案例代码一 阻塞队列
1.1.阻塞队列概念
概念
在多线程领域所谓阻塞在某些情况下会挂起线程即阻塞⼀旦条件满⾜被挂起的线程⼜ 会⾃动被唤醒。阻塞队列 是⼀个队列在数据结构中起的作⽤如下图 当队列是空的从队列中获取Take元素的操作将会被阻塞 当队列是满的从队列中添加Put元素的操作将会被阻塞 试图中空的队列中获取元素的线程将会被阻塞直到其他线程往空的队列插⼊新的元素 试图向已满的队列中添加新元素的线程将会被阻塞直到其他线程从队列中移除⼀个或多个元素 或者完全清空使队列变得空闲起来后并后续新增 好处阻塞队列不⽤⼿动控制什么时候该被阻塞什么时候该被唤醒简化了操作。 体系 Collection→ Queue→ BlockingQueue→七个阻塞队列实现类。 粗体标记的三个⽤得⽐较多许多消息中间件底层就是⽤它们实现的。 需要注意的是 LinkedBlockingQueue 虽然是有界的但有个巨坑其默认⼤⼩是 Integer.MAX_VALUE ⾼达21亿⼀般情况下内存早爆了在线程池的 ThreadPoolExecutor 有体现。 API 抛出异常是指 当队列满时再次插⼊会抛出异常 返回布尔是指当队列满时再次插⼊会返回false 阻塞是指当队列满时再次插⼊会被阻塞直到队列取出⼀个元素才能插⼊。 超时是指当⼀个时限过后才会插⼊或者取出。 1.2.阻塞队列API案例
1.2.1. ArrayBlockingQueue
1.2.1.1.抛出异常 代码
public class BlockingQueueDemo {public static void main(String[] args) throws InterruptedException {//定义容量为3的阻塞队列BlockingQueueString blockingQueue new ArrayBlockingQueueString(3);System.out.println(blockingQueue.add(a));System.out.println(blockingQueue.add(b));System.out.println(blockingQueue.add(c));System.out.println(blockingQueue.add(e));System.out.println(blockingQueue.element());System.out.println(blockingQueue.remove());System.out.println(blockingQueue.remove());System.out.println(blockingQueue.remove());System.out.println(blockingQueue.remove());}结果分析
当添加四个元素时抛出IllegalStateException: Queue full 当取出四个元素时抛出NoSuchElementException 1.2.1.2.返回布尔 代码 public static void main(String[] args) throws InterruptedException {//定义容量为3的阻塞队列BlockingQueueString blockingQueue new ArrayBlockingQueueString(3);System.out.println(blockingQueue.offer(a));System.out.println(blockingQueue.offer(b));System.out.println(blockingQueue.offer(c));System.out.println(blockingQueue.offer(e));System.out.println(blockingQueue.peek());System.out.println(blockingQueue.poll());System.out.println(blockingQueue.poll());System.out.println(blockingQueue.poll());System.out.println(blockingQueue.poll());}
结果 1.2.1.3.阻塞
代码 public static void main(String[] args) throws InterruptedException {//定义容量为3的阻塞队列BlockingQueueString blockingQueue new ArrayBlockingQueueString(3);blockingQueue.put(a);blockingQueue.put(b);blockingQueue.put(c);blockingQueue.put(d);System.out.println(blockingQueue.take());System.out.println(blockingQueue.take());System.out.println(blockingQueue.take());System.out.println(blockingQueue.take());}
结果
存储是容量已满 取得时候队列为空也会阻塞
1.2.1.4.超时
代码 public static void main(String[] args) throws InterruptedException {//定义容量为3的阻塞队列BlockingQueueString blockingQueue new ArrayBlockingQueueString(3);System.out.println(blockingQueue.offer(a,2L, TimeUnit.SECONDS));System.out.println(blockingQueue.offer(a,2L, TimeUnit.SECONDS));System.out.println(blockingQueue.offer(a,2L, TimeUnit.SECONDS));System.out.println(blockingQueue.offer(a,2L, TimeUnit.SECONDS));}结果
1.2.2.SynchronousQueue
队列只有⼀个元素如果想插⼊多个阻塞到队列元素取出后才能插⼊只能有⼀个“坑位”⽤⼀个 插⼀个详⻅SynchronousQueueDemo。 代码 public class SynchronousQueueDemo {public static void main(String[] args) {BlockingQueueString blockingQueuenew SynchronousQueueString();new Thread(()-{try {System.out.println(Thread.currentThread().getName()\t put 1);blockingQueue.put(1);System.out.println(Thread.currentThread().getName()\t put 2);blockingQueue.put(2);System.out.println(Thread.currentThread().getName()\t put 3);blockingQueue.put(3);} catch (InterruptedException e) {e.printStackTrace();}},AAA).start();new Thread(()-{try {try{ TimeUnit.SECONDS.sleep(5); }catch (InterruptedException e){ e.printStackTrace(); }System.out.println(Thread.currentThread().getName()\t take blockingQueue.take());try{ TimeUnit.SECONDS.sleep(5); }catch (InterruptedException e){ e.printStackTrace(); }System.out.println(Thread.currentThread().getName()\t take blockingQueue.take());try{ TimeUnit.SECONDS.sleep(5); }catch (InterruptedException e){ e.printStackTrace(); }System.out.println(Thread.currentThread().getName()\t takeblockingQueue.take());} catch (Exception e) {e.printStackTrace();}},BBB).start();}
}
结果
二 阻塞队列应用—生产者消费者
2.1.传统模式
传统模式使⽤ Synchronized来进⾏操作。
案例代码
/*** 题目现在两个线程可以操作初始值为零的一个变量* 实现一个线程对该变量加1一个线程对该变量-1* 实现交替来10轮变量初始值为0.* 1.高内聚低耦合前提下线程操作资源类* 2.判断/干活/通知* 3.防止虚假唤醒(判断只能用while不能用if)* 知识小总结多线程编程套路while判断新版写法*/
public class ProdConsumerDemo {public static void main(String[] args) {Aircondition aircondition new Aircondition();new Thread(() - {for (int i 1; i 10; i) {try {aircondition.increment();} catch (Exception e) {e.printStackTrace();}}}, A).start();new Thread(() - {for (int i 1; i 5; i) {try {aircondition.decrement();} catch (Exception e) {e.printStackTrace();}}}, B).start();}
}
class Aircondition {private int number 0;//⽼版写法public synchronized void increment() throws Exception {//1.判断if (number ! 0) {this.wait();}//2.⼲活number;System.out.println(Thread.currentThread().getName() \t number);//3通知this.notifyAll();}public synchronized void decrement() throws Exception {//1.判断if (number 0) {this.wait();}//2.⼲活number--;System.out.println(Thread.currentThread().getName() \t number);//3通知this.notifyAll();}
}结果 案例问题—防止虚假唤醒
上述我们只用了一个线程作为生产者一个线程作为消费者我们用多个来进行测试。防止虚假唤醒(判断只能用while不能用if) while循环与if判断 while是循环语句当满足条件时执行语句执行完循环以后再回来判断是否满足满足继续执行然后继续判断不满足直接执行下面的语句 if是判断语句满足条件就行执行完以后继续执行下面的语句不会再回来判断执行。 A线程生产包子B线程消费包子C线程生产包子D线程消费包子
public class ProdConsumerDemo {public static void main(String[] args) {Aircondition aircondition new Aircondition();new Thread(() - {for (int i 1; i 10; i) {try {aircondition.increment();} catch (Exception e) {e.printStackTrace();}}}, A).start();new Thread(() - {for (int i 1; i 5; i) {try {aircondition.decrement();} catch (Exception e) {e.printStackTrace();}}}, B).start();new Thread(() - {for (int i 1; i 10; i) {try {aircondition.increment();} catch (Exception e) {e.printStackTrace();}}}, C).start();new Thread(() - {for (int i 1; i 10; i) {try {aircondition.decrement();} catch (Exception e) {e.printStackTrace();}}}, D).start();}
}Aircondition
class Aircondition {private int number 0;//⽼版写法public synchronized void increment() throws Exception {//1.判断while (number ! 0) {this.wait();}//2.⼲活number;System.out.println(Thread.currentThread().getName() \t number);//3通知this.notifyAll();}public synchronized void decrement() throws Exception {//1.判断while (number 0) {this.wait();}//2.⼲活number--;System.out.println(Thread.currentThread().getName() \t number);//3通知this.notifyAll();}
}2.2.⽣产者消费者防⽌虚假唤醒
2.2.1 新版⽣产者消费者写法 ReentrantLock.Condition
Synchronized用在多线程中太重了在高并发场景使用lock方式更加合适。所以我们使用lock来加锁和解锁。对应的等待和唤醒线程方法也换成java.util.concurrent.locks下面的newCondition方法。
案例代码
class Aircondition{private int number 0;private Lock lock new ReentrantLock();private Condition condition lock.newCondition();//新版写法public void increment() throws Exception{lock.lock();try{//1.判断while (number ! 0){condition.await();}//2.干活number;System.out.println(Thread.currentThread().getName()\tnumber);//3通知condition.signalAll();} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();}}public void decrement() throws Exception{lock.lock();try{//1.判断while (number 0){condition.await();}//2.干活number--;System.out.println(Thread.currentThread().getName()\tnumber);//3通知condition.signalAll();} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();}}
}public class ProdConsumerDemo {public static void main(String[] args) {Aircondition aircondition new Aircondition();new Thread(() - {for (int i 1; i 10; i) {try {aircondition.increment();} catch (Exception e) {e.printStackTrace();}}}, A).start();new Thread(() - {for (int i 1; i 5; i) {try {aircondition.decrement();} catch (Exception e) {e.printStackTrace();}}}, B).start();new Thread(() - {for (int i 1; i 10; i) {try {aircondition.increment();} catch (Exception e) {e.printStackTrace();}}}, C).start();new Thread(() - {for (int i 1; i 10; i) {try {aircondition.decrement();} catch (Exception e) {e.printStackTrace();}}}, D).start();}
}2.2.2精准通知顺序访问
现在有这样一个需求 备注多线程之间按顺序调用实现A-B-C 三个线程启动要求如下 A打印5次B打印10次C打印15次 接着 A打印5次B打印10次C打印15次 来10轮 1.高内聚低耦合前提下线程操作资源类 2.判断/干活/通知 3.多线程交互中防止虚假唤醒(判断只能用while不能用if) 4.标志位 案例代码
class ShareData{private int number 1;//A:1,B:2,C:3private Lock lock new ReentrantLock();private Condition c1 lock.newCondition();private Condition c2 lock.newCondition();private Condition c3 lock.newCondition();public void printc1(){lock.lock();try {//1.判断while (number ! 1){c1.await();}//2.干活for (int i 1; i 5; i) {System.out.println(Thread.currentThread().getName()\ti);}//3.通知number 2;//通知第2个c2.signal();} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();}}public void printc2(){lock.lock();try {//1.判断while (number ! 2){c2.await();}//2.干活for (int i 1; i 10; i) {System.out.println(Thread.currentThread().getName()\ti);}//3.通知number 3;//如何通知第3个c3.signal();} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();}}public void printc3(){lock.lock();try {//1.判断while (number ! 3){c3.await();}//2.干活for (int i 1; i 15; i) {System.out.println(Thread.currentThread().getName()\ti);}//3.通知number 1;//如何通知第1个c1.signal();} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();}}
}/*** 备注多线程之间按顺序调用实现A-B-C* 三个线程启动要求如下* A打印5次B打印10次C打印15次* 接着* A打印5次B打印10次C打印15次* 来10轮* 1.高内聚低耦合前提下线程操作资源类* 2.判断/干活/通知* 3.多线程交互中防止虚假唤醒(判断只能用while不能用if)* 4.标志位*/
public class ConditionDemo {public static void main(String[] args) {ShareData shareData new ShareData();new Thread(()-{for (int i 1; i 10; i) {shareData.printc1();}},A).start();new Thread(()-{for (int i 1; i 10; i) {shareData.printc2();}},B).start();new Thread(()-{for (int i 1; i 10; i) {shareData.printc3();}},C).start();}
}
结果 结果分析
2.3.Synchronized和Lock的区别
synchronized关键字和 java.util.concurrent.locks.Lock都能加锁两者有什么区别呢
原始构成 sync是JVM层⾯的底层通过monitorenter和monitorexit来实现的。 Lock是 JDK API层⾯的。 sync⼀个enter会有两个exit⼀个是正常退出⼀个是异常退出使⽤⽅法 sync不需要⼿动释放锁⽽ Lock需要⼿动释放。是否可中断 sync不可中断除⾮抛出异常或者正常运⾏完成。 Lock是可中断的通过调 ⽤ interrupt()⽅法。是否为公平锁 sync只能是⾮公平锁⽽ Lock既能是公平锁⼜能是⾮公平锁。绑定多个条件 sync不能只能随机唤醒。⽽ Lock可以通过 Condition来绑定多个条件精确唤醒。
2.4. 阻塞队列模式⽣产者消费者
为什么需要BlockingQueue 好处是我们不需要关⼼什么时候需要阻塞线程什么时候需要唤醒线程因为这⼀切BlockingQueue都 给你⼀⼿包办好了使⽤阻塞队列 后就不需要⼿动加锁了。 在Concurrent包发布以前在多线程环境下我们每个程序员都必须去⾃⼰控制这些细节尤其还要兼 顾效率和线程安全⽽这会给我们的程序带来不⼩的复杂度。 案例代码
public class ProdConsBlockQueueDemo {public static void main(String[] args) {MyResource myResource new MyResource(new ArrayBlockingQueue(5));new Thread(() - {System.out.println(Thread.currentThread().getName() \t生产线程启动);try {myResource.myProd();} catch (Exception e) {e.printStackTrace();}}, prod).start();new Thread(() - {System.out.println(Thread.currentThread().getName() \t生产线程启动);try {myResource.myProd();} catch (Exception e) {e.printStackTrace();}}, prod-2).start();new Thread(() - {System.out.println(Thread.currentThread().getName() \t消费线程启动);try {myResource.myCons();} catch (Exception e) {e.printStackTrace();}}, cons).start();new Thread(() - {System.out.println(Thread.currentThread().getName() \t消费线程启动);try {myResource.myCons();} catch (Exception e) {e.printStackTrace();}}, cons-2).start();try {TimeUnit.SECONDS.sleep(5);} catch (Exception e) {e.printStackTrace();}System.out.println(5秒钟后叫停);myResource.stop();}
}class MyResource {// 定义成volatile类型线程间可见 默认开启进行生产消费private volatile boolean FLAG true;//定义原子Integer类型保证原子性private AtomicInteger atomicInteger new AtomicInteger();private BlockingQueueString blockingQueue;public MyResource(BlockingQueueString blockingQueue) {this.blockingQueue blockingQueue;System.out.println(blockingQueue.getClass().getName());}public void myProd() throws Exception {String data null;boolean retValue;while (FLAG) {data atomicInteger.incrementAndGet() ;//iretValue blockingQueue.offer(data, 2L, TimeUnit.SECONDS);if (retValue) {System.out.println(Thread.currentThread().getName() \t 插入队列 data 成功);} else {System.out.println(Thread.currentThread().getName() \t 插入队列 data 失败);}TimeUnit.SECONDS.sleep(1);}System.out.println(Thread.currentThread().getName() \t老板叫停了FLAG已更新为false停止生产);}public void myCons() throws Exception {String res;while (FLAG) {res blockingQueue.poll(2L, TimeUnit.SECONDS);if (null res || .equals(res)) {// FLAG false;System.out.println(Thread.currentThread().getName() \t超过2秒钟没有消费退出消费);return;}System.out.println(Thread.currentThread().getName() \t\t消费队列 res 成功);}}public void stop() {this.FLAG false;}
}
结果分析
