阻塞队列和定时器的使用

阻塞队列

谈到队列,大家就能想到队列的先进先出原则,但有些特殊的队列,虽然也是先进先出的,但是带有阻塞功能,我们把这种队列叫做阻塞队列.

★如果队列为空,执行出队操作就会阻塞,阻塞到另外一个线程往队列里添加元素(队列不为空)为止.

★如果队列满了,执行入队操作时,也会阻塞,阻塞到另外一个线程从队列里取出元素(队列不满).

想要了解阻塞队列,先了解生产者-消费者模型,这个模型给我们程序带来了非常重要的好处

1.生产者-消费者模型

在一个餐馆中，有多名厨师（生产者）负责烹饪菜品，并将其放在共享的出菜区（缓冲区）上。而服务员（消费者）则从出菜区上取出菜品并端给顾客。这里的厨师可以看作是生产者，出菜区是缓冲区，服务员是消费者。

1. 实现了生产者和消费者之间的 "解耦"

我们写代码要追求低耦合,所谓解耦操作,就是降低耦合的过程.

        比如过年一家人一起包饺子. 一般都是有明确分工 , 比如一个人负责擀饺子皮 , 其他人负责包 . 擀饺 子皮的人就是 "生产者", 包饺子的人就是 " 消费者 ".

        擀饺子皮的人不关心包饺子的人是谁( 能包就行 , 无论是手工包 , 借助工具 , 还是机器包 ), 包饺子的人 也不关心擀饺子皮的人是谁( 有饺子皮就行 , 无论是用擀面杖擀的 , 还是拿罐头瓶擀 , 还是直接从超市买的)

上述操作生产者和消费者没有直接联系,而是通过阻塞队列相互关联,生产者和消费者有一方出现了问题,不会影响对方的操作,这样操作就降低了耦合.

2.可以"削峰填谷",保证系统的稳定性

阻塞队列就相当于一个缓冲区,平衡了生产者消费者的处理能力

比如在 " 秒杀 " 场景下 , 服务器同一时刻可能会收到大量的支付请求 . 如果直接处理这些支付请求 , 服务器可能扛不住( 每个支付请求的处理都需要比较复杂的流程 ). 这个时候就可以把这些请求都放 到一个阻塞队列中, 然后再由消费者线程慢慢的来处理每个支付请求 .

这样做可以有效进行 " 削峰 ", 防止服务器被突然到来的一波请求直接冲垮

2.标准库中的阻塞队列

java标准库中内置了阻塞队列,在使用的时候直接使用标准库中的就行

1.BlockingDeque是一个接口

2.真正实现BlockingDeque的类有三种

3.带有阻塞功能的方法.put()方法用于阻塞队列的入队,take()方法用于阻塞队列的出队.在使用take的时候需要处理异常

public static void main(String[] args) {BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<>();//创建两个线程,来作为生产者和消费者Thread customer = new Thread(() -> {while (true) {try {Integer result = blockingQueue.take();System.out.println("消费元素: " + result);} catch (InterruptedException e) {e.printStackTrace();}}});customer.start();Thread producer = new Thread( () -> {int count = 0;while (true) {try {blockingQueue.put(count);System.out.println("生产元素: " + count );count++;Thread.sleep(500);} catch (InterruptedException e) {e.printStackTrace();}}});producer.start();}

上面执行的结果是

 上面代码先执行消费元素是因为当customer线程执行到blockingQueue,take()时发生了阻塞,等到producer线程执行blockingQueue.put()后,两个线程并发执行,谁先打印都有可能.

4. 不带阻塞功能的方法 入队列 offer(), 出队列 poll(),  取队首元素 peek().

3.模拟实现阻塞队列

实现阻塞队列的前提是我们要先实现一个普通的队列,实现普通队列的方法有两种,一个是基于链表的队列,采用头删法和尾插法实现一个阻塞队列,另外一个方法是基于数组的循环队列,两个方法的具体实现在下面这篇文章里面.

循环队列的实现.​​​​​​​

要模拟实现阻塞队列,前提是在多线程的基础上,通过synchronized关键字对put()和take()方法中的代码进行加锁,然后通过wait()和notify()方法对插入元素和取出元素进行阻塞等待.具体方法是,当插入元素时如果队列满了,就产生阻塞等待,直到有元素取出后唤醒put中的wait().当取出元素的时候,如果队列为空,让take()方法产生阻塞等待,只到有元素插入后,唤醒take()中的wait().

class MyBlockingQueue {private int[] item = new int[1000];private int head = 0;private int tail = 0;private int size = 0;//插入元素public void put(int value) throws InterruptedException {synchronized (this) {while(size == item.length) {//队列满了,产生阻塞等待this.wait();}item[tail] = value;tail++;if(tail >= item.length) {tail = 0;}size++;//唤醒take中的wait()this.notify();}}//取出元素public Integer take() throws InterruptedException {int result = 0;synchronized (this) {while(size == 0) {//队列为空,产生阻塞等待this.wait();}result = item[head];head++;if(head >= item.length) {head = 0;}size--;//唤醒put中的wait()this.notify();}return result;}
}

上面代码中put()和take()不会同时产生阻塞等待,因为我们是在同一个对象中,一个队列不可能同时满足队列为空和队列为满.

定时器

定时器就好比是我们生活中的闹钟,当我们被早八的闹钟叫醒后想在床上在躺十分钟,我们设置闹钟就可以让十分钟再响,而我们的计时器具有类似的功能,是指定一段之间后执行一段代码.

在Java的标准库中有一个Timer类,Timer类的核心方法是schedule()

schedule()中有连个参数,第一个参数指定定时器要执行的任务,第二个参数指定多长时间后执行这个任务

观察代码结果发现,在 "程序启动" 打印后3000ms才打印  "运行定时器任务".

模拟实现定时器

主要实现下面几种功能

1.让被注册的任务,能在指定时间后执行.为了实现这个功能可以单独在定时器内部建一个线程,让这个线程周期性的扫描,判断任务时间是否到了,如果到了,就执行,若没动,就继续等待.这n个任务,可以用优先级队列来保存,按照指定时间小的优先级高,就放在队头,此时扫描线程,就只需要扫描队头元素,对头元素时间没到的话,其他时间更不可能到.

在优先级队列中,我们是基于时间进行比较的,可以实现 Comparable接口来进行比较

在扫描线程中会出现下面的bug.

2.一个定时器可以指定N个任务,N个任务会按照指定时间按顺序执行.

具体实现代码

class MyTask implements Comparable<MyTask>{//要执行的任务private Runnable runnable;//任务在什么时间执行(用毫秒时间戳表示)private long time;public MyTask(Runnable runnable, long time) {this.runnable = runnable;this.time = time;}//获取当前任务的时间public long getTime() {return time;}//执行任务public void run() {runnable.run();}@Overridepublic int compareTo(MyTask o) {return (int) (this.time - o.time);}
}//定时器
class MyTimer {//扫描线程private Thread t = null;//有一个阻塞优先级队列private PriorityBlockingQueue<MyTask> queue = new PriorityBlockingQueue<>();public MyTimer() {t = new Thread(() -> {while (true) {try {synchronized (this) {MyTask myTask = queue.take();//取出队头元素观察判断是否到时间了long curTime = System.currentTimeMillis();if(curTime < myTask.getTime()) {//时间没到放回队列queue.put(myTask);} else {//时间到了,执行任务myTask.run();this.wait(myTask.getTime() - curTime);}}} catch (InterruptedException e) {e.printStackTrace();}}} );t.start();}public void schedule (Runnable runnable,long after) {//第一个参数 是指定的任务// 第二个参数是在多少毫秒后执行MyTask myTask = new MyTask(runnable,System.currentTimeMillis()+ after);//把任务放到队列里queue.put(myTask);synchronized (this) {this.notify();}}