山西营销型企业网站开发,网络舆情分析报告,产品设计方案怎么写,免费领取一个微信号什么是阻塞队列
相比于一般的队列#xff0c;有两个特点 1.线程安全 2.带有阻塞功能 1#xff09;队伍为空时#xff0c;出队列就会出现阻塞#xff0c;阻塞到其他线程入队列为止 2#xff09;队伍为满时#xff0c;入队列就会出现阻塞#xff0c;阻塞到其他线程出队列…什么是阻塞队列
相比于一般的队列有两个特点 1.线程安全 2.带有阻塞功能 1队伍为空时出队列就会出现阻塞阻塞到其他线程入队列为止 2队伍为满时入队列就会出现阻塞阻塞到其他线程出队列为止
常用于生产者消费者模型 作用 1.解耦合 2.削峰填谷
使用阻塞队列
public class Test12 {public static void main(String[] args) throws InterruptedException {BlockingQueueString queue new ArrayBlockingQueue(100);queue.put(qqq);String elem queue.take();System.out.println(elem elem);elem queue.take();System.out.println(elem elem);}
}
运行结果 不会结束运行一直在等待。
使用put和offer一样的都是入队列但是put是带有阻塞功能offer没有带阻塞功能队满了就会返回结果 take方法用来出队列也是带有阻塞功能
实现阻塞队列
1实现普通队列
class MyBlockingQueue{private String[] elems null;private int size 0;private int head 0;private int tail 0;public MyBlockingQueue(int capacity){elems new String[capacity];}public void put(String elem) throws InterruptedException {if (size elems.length){//阻塞功能}elems[tail] elem;tail;if (tail elems.length){tail 0;}size;}public String take() throws InterruptedException {if (size 0){//阻塞功能}String elem null;elem elems[head];head;if (head elems.length){head 0;}size--;return elem;}}}2加上线程安全
class MyBlockingQueue{private String[] elems null;private int size 0;private int head 0;private int tail 0;private Object locker new Object();public MyBlockingQueue(int capacity){elems new String[capacity];}public void put(String elem) throws InterruptedException {synchronized(locker){if (size elems.length){}elems[tail] elem;tail;if (tail elems.length){tail 0;}size;}}public String take() throws InterruptedException {synchronized(locker){if (size 0){}String elem null;elem elems[head];head;if (head elems.length){head 0;}size--;return elem;}}}3加上阻塞功能
class MyBlockingQueue{private String[] elems null;private int size 0;private int head 0;private int tail 0;private Object locker new Object();public MyBlockingQueue(int capacity){elems new String[capacity];}public void put(String elem) throws InterruptedException {synchronized(locker){while (size elems.length){locker.wait();}elems[tail] elem;tail;if (tail elems.length){tail 0;}size;locker.notify();}}public String take() throws InterruptedException {synchronized(locker){while (size 0){locker.wait();}String elem null;elem elems[head];head;if (head elems.length){head 0;}size--;locker.notify();return elem;}}}代码解释 最终代码将if改成了while因为if只能判定一次如果出现以下情况就会出bug线程A线程B都执行到了put中的wait因为队列已满而停止运行线程C出队列唤醒了线程A线程A继续入队列入队列后就会notify导致唤醒了线程B而此时队列已满无法进行入队操作就出现了bug所以就使用whilewait之前判定一次唤醒之后再进行一次判定相当于多做一步确定操作
简单的生产者消费者模型
class MyBlockingQueue{private String[] elems null;private int size 0;private int head 0;private int tail 0;private Object locker new Object();public MyBlockingQueue(int capacity){elems new String[capacity];}public void put(String elem) throws InterruptedException {synchronized(locker){while (size elems.length){locker.wait();}elems[tail] elem;tail;if (tail elems.length){tail 0;}size;locker.notify();}}public String take() throws InterruptedException {synchronized(locker){while (size 0){locker.wait();}String elem null;elem elems[head];head;if (head elems.length){head 0;}size--;locker.notify();return elem;}}}
public class Test11 {public static void main(String[] args) throws InterruptedException {MyBlockingQueue myBlockingQueue new MyBlockingQueue(100);Thread t1 new Thread(()-{int n 1;while(true){try {myBlockingQueue.put(n );System.out.println(生产元素 n);n;} catch (InterruptedException e) {throw new RuntimeException(e);}}});Thread t2 new Thread(()-{while(true){try {String n myBlockingQueue.take();System.out.println(消费元素 n);Thread.sleep(500);} catch (InterruptedException e) {throw new RuntimeException(e);}}});t1.start();t2.start();}
}
