好了,话不多说,直接进入今天的主题,本文的主要内容如下图所示:
顾明思议,我们把需要延迟执行的任务叫做延迟任务。
延迟任务的使用场景有以下这些:
等事件都需要使用延迟任务。
延迟任务实现的关键是在某个时间节点执行某个任务。基于这个信息我们可以想到实现延迟任务的手段有以下两个:
而通过 JDK 实现延迟任务我们能想到的关键词是:DelayQueue、ScheduledExecutorService,而第三方提供的延迟任务执行方法就有很多了,例如:Redis、Netty、MQ 等手段。
下面我们将结合代码来讲解每种延迟任务的具体实现。
此方式我们需要开启一个无限循环一直扫描任务,然后使用一个 Map 集合用来存储任务和延迟执行的时间,实现代码如下:
import java.time.Instant;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
/**
* 延迟任务执行方法汇总
*/
public class DelayTaskExample {
// 存放定时任务
private static Map<String, Long> _TaskMap = new HashMap<>();
public static void main(String[] args) {
System.out.println("程序启动时间:" + LocalDateTime.now());
// 添加定时任务
_TaskMap.put("task-1", Instant.now().plusSeconds(3).toEpochMilli()); // 延迟 3s
// 调用无限循环实现延迟任务
loopTask();
}
/**
* 无限循环实现延迟任务
*/
public static void loopTask() {
Long itemLong = 0L;
while (true) {
Iterator it = _TaskMap.entrySet().iterator();
while (it.hasNext()) {
Map.Entry entry = (Map.Entry) it.next();
itemLong = (Long) entry.getValue();
// 有任务需要执行
if (Instant.now().toEpochMilli() >= itemLong) {
// 延迟任务,业务逻辑执行
System.out.println("执行任务:" + entry.getKey() +
" ,执行时间:" + LocalDateTime.now());
// 删除任务
_TaskMap.remove(entry.getKey());
}
}
}
}
}
以上程序执行的结果为:
程序启动时间:2020-04-12T18:51:28.188
执行任务:task-1 ,执行时间:2020-04-12T18:51:31.189
可以看出任务延迟了 3s 钟执行了,符合我们的预期。
Java API 提供了两种实现延迟任务的方法:DelayQueue 和 ScheduledExecutorService。
我们可以使用 ScheduledExecutorService 来以固定的频率一直执行任务,实现代码如下:
public class DelayTaskExample {
public static void main(String[] args) {
System.out.println("程序启动时间:" + LocalDateTime.now());
scheduledExecutorServiceTask();
}
/**
* ScheduledExecutorService 实现固定频率一直循环执行任务
*/
public static void scheduledExecutorServiceTask() {
ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
executor.scheduleWithFixedDelay(
new Runnable() {
@Override
public void run() {
// 执行任务的业务代码
System.out.println("执行任务" +
" ,执行时间:" + LocalDateTime.now());
}
},
2, // 初次执行间隔
2, // 2s 执行一次
TimeUnit.SECONDS);
}
}
以上程序执行的结果为:
程序启动时间:2020-04-12T21:28:10.416
执行任务 ,执行时间:2020-04-12T21:28:12.421
执行任务 ,执行时间:2020-04-12T21:28:14.422
......
可以看出使用 ScheduledExecutorService#scheduleWithFixedDelay(...) 方法之后,会以某个频率一直循环执行延迟任务。
DelayQueue 是一个支持延时获取元素的无界阻塞队列,队列中的元素必须实现 Delayed 接口,并重写 getDelay(TimeUnit) 和 compareTo(Delayed) 方法,DelayQueue 实现延迟队列的完整代码如下:
public class DelayTest {
public static void main(String[] args) throws InterruptedException {
DelayQueue delayQueue = new DelayQueue();
// 添加延迟任务
delayQueue.put(new DelayElement(1000));
delayQueue.put(new DelayElement(3000));
delayQueue.put(new DelayElement(5000));
System.out.println("开始时间:" + DateFormat.getDateTimeInstance().format(new Date()));
while (!delayQueue.isEmpty()){
// 执行延迟任务
System.out.println(delayQueue.take());
}
System.out.println("结束时间:" + DateFormat.getDateTimeInstance().format(new Date()));
}
static class DelayElement implements Delayed {
// 延迟截止时间(单面:毫秒)
long delayTime = System.currentTimeMillis();
public DelayElement(long delayTime) {
this.delayTime = (this.delayTime + delayTime);
}
@Override
// 获取剩余时间
public long getDelay(TimeUnit unit) {
return unit.convert(delayTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
@Override
// 队列里元素的排序依据
public int compareTo(Delayed o) {
if (this.getDelay(TimeUnit.MILLISECONDS) > o.getDelay(TimeUnit.MILLISECONDS)) {
return 1;
} else if (this.getDelay(TimeUnit.MILLISECONDS) < o.getDelay(TimeUnit.MILLISECONDS)) {
return -1;
} else {
return 0;
}
}
@Override
public String toString() {
return DateFormat.getDateTimeInstance().format(new Date(delayTime));
}
}
}
开始时间:2020-4-12 20:40:38
2020-4-12 20:40:39
2020-4-12 20:40:41
2020-4-12 20:40:43
结束时间:2020-4-12 20:40:43
使用 Redis 实现延迟任务的方法大体可分为两类:通过 zset 数据判断的方式,和通过键空间通知的方式。
我们借助 zset 数据类型,把延迟任务存储在此数据集合中,然后在开启一个无线循环查询当前时间的所有任务进行消费,实现代码如下(需要借助 Jedis 框架):
import redis.clients.jedis.Jedis;
import utils.JedisUtils;
import java.time.Instant;
import java.util.Set;
public class DelayQueueExample {
// zset key
private static final String _KEY = "myDelayQueue";
public static void main(String[] args) throws InterruptedException {
Jedis jedis = JedisUtils.getJedis();
// 延迟 30s 执行(30s 后的时间)
long delayTime = Instant.now().plusSeconds(30).getEpochSecond();
jedis.zadd(_KEY, delayTime, "order_1");
// 继续添加测试数据
jedis.zadd(_KEY, Instant.now().plusSeconds(2).getEpochSecond(), "order_2");
jedis.zadd(_KEY, Instant.now().plusSeconds(2).getEpochSecond(), "order_3");
jedis.zadd(_KEY, Instant.now().plusSeconds(7).getEpochSecond(), "order_4");
jedis.zadd(_KEY, Instant.now().plusSeconds(10).getEpochSecond(), "order_5");
// 开启延迟队列
doDelayQueue(jedis);
}
/**
* 延迟队列消费
* @param jedis Redis 客户端
*/
public static void doDelayQueue(Jedis jedis) throws InterruptedException {
while (true) {
// 当前时间
Instant nowInstant = Instant.now();
long lastSecond = nowInstant.plusSeconds(-1).getEpochSecond(); // 上一秒时间
long nowSecond = nowInstant.getEpochSecond();
// 查询当前时间的所有任务
Set<String> data = jedis.zrangeByScore(_KEY, lastSecond, nowSecond);
for (String item : data) {
// 消费任务
System.out.println("消费:" + item);
}
// 删除已经执行的任务
jedis.zremrangeByScore(_KEY, lastSecond, nowSecond);
Thread.sleep(1000); // 每秒轮询一次
}
}
}
默认情况下 Redis 服务器端是不开启键空间通知的,需要我们通过 config set notify-keyspace-events Ex 的命令手动开启,开启键空间通知后,我们就可以拿到每个键值过期的事件,我们利用这个机制实现了给每个人开启一个定时任务的功能,实现代码如下:
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;
import utils.JedisUtils;
public class TaskExample {
public static final String _TOPIC = "__keyevent@0__:expired"; // 订阅频道名称
public static void main(String[] args) {
Jedis jedis = JedisUtils.getJedis();
// 执行定时任务
doTask(jedis);
}
/**
* 订阅过期消息,执行定时任务
* @param jedis Redis 客户端
*/
public static void doTask(Jedis jedis) {
// 订阅过期消息
jedis.psubscribe(new JedisPubSub() {
@Override
public void onpmessage(String pattern, String channel, String message) {
// 接收到消息,执行定时任务
System.out.println("收到消息:" + message);
}
}, _TOPIC);
}
}
Netty 是由 JBOSS 提供的一个 Java 开源框架,它是一个基于 NIO 的客户、服务器端的编程框架,使用 Netty 可以确保你快速和简单的开发出一个网络应用,例如实现了某种协议的客户、服务端应用。Netty 相当于简化和流线化了网络应用的编程开发过程,例如:基于 TCP 和 UDP 的 socket 服务开发。
可以使用 Netty 提供的工具类 HashedWheelTimer 来实现延迟任务,实现代码如下。
首先在项目中添加 Netty 引用,配置如下:
<!-- https://mvnrepository.com/artifact/io.netty/netty-common -->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-common</artifactId>
<version>4.1.48.Final</version>
</dependency>
Netty 实现的完整代码如下:
public class DelayTaskExample {
public static void main(String[] args) {
System.out.println("程序启动时间:" + LocalDateTime.now());
NettyTask();
}
/**
* 基于 Netty 的延迟任务
*/
private static void NettyTask() {
// 创建延迟任务实例
HashedWheelTimer timer = new HashedWheelTimer(3, // 时间间隔
TimeUnit.SECONDS,
100); // 时间轮中的槽数
// 创建一个任务
TimerTask task = new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
System.out.println("执行任务" +
" ,执行时间:" + LocalDateTime.now());
}
};
// 将任务添加到延迟队列中
timer.newTimeout(task, 0, TimeUnit.SECONDS);
}
}
以上程序执行的结果为:
程序启动时间:2020-04-13T10:16:23.033
执行任务 ,执行时间:2020-04-13T10:16:26.118
HashedWheelTimer 是使用定时轮实现的,定时轮其实就是一种环型的数据结构,可以把它想象成一个时钟,分成了许多格子,每个格子代表一定的时间,在这个格子上用一个链表来保存要执行的超时任务,同时有一个指针一格一格的走,走到那个格子时就执行格子对应的延迟任务,如下图所示:
以上的图片可以理解为,时间轮大小为 8,某个时间转一格(例如 1s),每格指向一个链表,保存着待执行的任务。
如果专门开启一个 MQ 中间件来执行延迟任务,就有点杀鸡用宰牛刀般的奢侈了,不过已经有了 MQ 环境的话,用它来实现延迟任务的话,还是可取的。
几乎所有的 MQ 中间件都可以实现延迟任务,在这里更准确的叫法应该叫延队列。本文就使用 RabbitMQ 为例,来看它是如何实现延迟任务的。
RabbitMQ 实现延迟队列的方式有两种:
注意:延迟插件 rabbitmq-delayed-message-exchange 是在 RabbitMQ 3.5.7 及以上的版本才支持的,依赖 Erlang/OPT 18.0 及以上运行环境。
由于使用死信交换器比较麻烦,所以推荐使用第二种实现方式 rabbitmq-delayed-message-exchange 插件的方式实现延迟队列的功能。
首先,我们需要下载并安装 rabbitmq-delayed-message-exchange 插件,下载地址:http://www.rabbitmq.com/community-plugins.html
选择相应的对应的版本进行下载,然后拷贝到 RabbitMQ 服务器目录,使用命令 rabbitmq-plugins enable rabbitmq_delayed_message_exchange 开启插件,在使用命令 rabbitmq-plugins list 查询安装的所有插件,安装成功如下图所示:
最后重启 RabbitMQ 服务,使插件生效。
首先,我们先要配置消息队列,实现代码如下:
import com.example.rabbitmq.mq.DirectConfig;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class DelayedConfig {
final static String QUEUE_NAME = "delayed.goods.order";
final static String EXCHANGE_NAME = "delayedec";
@Bean
public Queue queue() {
return new Queue(DelayedConfig.QUEUE_NAME);
}
// 配置默认的交换机
@Bean
CustomExchange customExchange() {
Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct");
//参数二为类型:必须是x-delayed-message
return new CustomExchange(DelayedConfig.EXCHANGE_NAME, "x-delayed-message", true, false, args);
}
// 绑定队列到交换器
@Bean
Binding binding(Queue queue, CustomExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(DelayedConfig.QUEUE_NAME).noargs();
}
}
然后添加增加消息的代码,具体实现如下:
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.text.SimpleDateFormat;
import java.util.Date;
@Component
public class DelayedSender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send(String msg) {
SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
System.out.println("发送时间:" + sf.format(new Date()));
rabbitTemplate.convertAndSend(DelayedConfig.EXCHANGE_NAME, DelayedConfig.QUEUE_NAME, msg, new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setHeader("x-delay", 3000);
return message;
}
});
}
}
再添加消费消息的代码:
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.text.SimpleDateFormat;
import java.util.Date;
@Component
@RabbitListener(queues = "delayed.goods.order")
public class DelayedReceiver {
@RabbitHandler
public void process(String msg) {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
System.out.println("接收时间:" + sdf.format(new Date()));
System.out.println("消息内容:" + msg);
}
}
最后,我们使用代码测试一下:
import com.example.rabbitmq.RabbitmqApplication;
import com.example.rabbitmq.mq.delayed.DelayedSender;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import java.text.SimpleDateFormat;
import java.util.Date;
@RunWith(SpringRunner.class)
@SpringBootTest
public class DelayedTest {
@Autowired
private DelayedSender sender;
@Test
public void Test() throws InterruptedException {
SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd");
sender.send("Hi Admin.");
Thread.sleep(5 * 1000); //等待接收程序执行之后,再退出测试
}
}
以上程序的执行结果如下:
发送时间:2020-04-13 20:47:51
接收时间:2020-04-13 20:47:54
消息内容:Hi Admin.
从结果可以看出,以上程序执行符合延迟任务的实现预期。
如果你使用的是 Spring 或 SpringBoot 的项目的话,可以使用借助 Scheduled 来实现,本文将使用 SpringBoot 项目来演示 Scheduled 的实现,实现我们需要声明开启 Scheduled,实现代码如下:
@SpringBootApplication
@EnableScheduling
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
}
然后添加延迟任务,实现代码如下:
@Component
public class ScheduleJobs {
@Scheduled(fixedDelay = 2 * 1000)
public void fixedDelayJob() throws InterruptedException {
System.out.println("任务执行,时间:" + LocalDateTime.now());
}
}
此时当我们启动项目之后就可以看到任务以延迟了 2s 的形式一直循环执行,结果如下:
任务执行,时间:2020-04-13T14:07:53.349
任务执行,时间:2020-04-13T14:07:55.350
任务执行,时间:2020-04-13T14:07:57.351
...
我们也可以使用 Corn 表达式来定义任务执行的频率,例如使用 @Scheduled(cron = "0/4 * * * * ?") 。
Quartz 是一款功能强大的任务调度器,可以实现较为复杂的调度功能,它还支持分布式的任务调度。
我们使用 Quartz 来实现一个延迟任务,首先定义一个执行任务代码如下:
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.scheduling.quartz.QuartzJobBean;
import java.time.LocalDateTime;
public class SampleJob extends QuartzJobBean {
@Override
protected void executeInternal(JobExecutionContext jobExecutionContext)
throws JobExecutionException {
System.out.println("任务执行,时间:" + LocalDateTime.now());
}
}
在定义一个 JobDetail 和 Trigger 实现代码如下:
import org.quartz.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class SampleScheduler {
@Bean
public JobDetail sampleJobDetail() {
return JobBuilder.newJob(SampleJob.class).withIdentity("sampleJob")
.storeDurably().build();
}
@Bean
public Trigger sampleJobTrigger() {
// 3s 后执行
SimpleScheduleBuilder scheduleBuilder =
SimpleScheduleBuilder.simpleSchedule().withIntervalInSeconds(3).withRepeatCount(1);
return TriggerBuilder.newTrigger().forJob(sampleJobDetail()).withIdentity("sampleTrigger")
.withSchedule(scheduleBuilder).build();
}
}
最后在 SpringBoot 项目启动之后开启延迟任务,实现代码如下:
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;
/**
* SpringBoot 项目启动后执行
*/
public class MyStartupRunner implements CommandLineRunner {
@Autowired
private SchedulerFactoryBean schedulerFactoryBean;
@Autowired
private SampleScheduler sampleScheduler;
@Override
public void run(String... args) throws Exception {
// 启动定时任务
schedulerFactoryBean.getScheduler().scheduleJob(
sampleScheduler.sampleJobTrigger());
}
}
以上程序的执行结果如下:
从结果可以看出在项目启动 3s 之后执行了延迟任务。
本文讲了延迟任务的使用场景,以及延迟任务的 10 种实现方式:
来源: Java中文社群
诸如tomcat等web服务器中间件简化了我们web的开发成本,但有时候我们或许并不需要这么一个完备的服务器,只是希望做一个简单地处理或者做特殊用途的服务器。
在以往的 Tomcat 项目中,一直习惯用 Ant 打包,使用 build.xml 配置,通过 ant -buildfile 的方式在机器上执行定时任务。虽然 Spring 本身支持定时任务,但都是服务一直运行时支持。
首先引入依赖,既然要进行智能联想,那么我们需要为提供联想的数据建立一个联想索引(而不是使用原来的数据索引),既然要建立索引,那么我们需要知道建立索引的数据来源。我们使用一个扩展自InputIterator的类来定义数据来源
本文是基于Java 8的HashMap进行分析,主要是介绍HashMap中的成员变量和类变量的用途,以及分析HashMap的数据结构。在HashMap中存在多个成员变量和类变量,搞清楚它们的用途有助于我们更深入了解HashMap,下面是它们的介绍:
Spring默认使用Jackson处理json数据。实际开发中,在业界中,使用非常受欢迎的fastjson来接受json数据。创建一个项目,在web目录下新建一个assets/js目录,加入jquery和json2的js文件,在lib下加入fastjson的jar文件。
求两个IP地址之间的IP个数,例如192.18.16.1~192.18.16.5,2001:DB8:0000:0023:0008:0800:200C:417C~2001:DB8:0:23:8:800:200C:417D之间的IP个数?
JSP和JSF这两种技术都基于Java,主要用于基于Web的应用程序。那么它们之间有什么区别?下面本篇文章就来给大家简单比较一下JSP和JSF,介绍JSP和JSF之间的区别有哪些,希望对大家有所帮助。
Java 堆空间:发生频率:5颗星造成原因1、无法在 Java 堆中分配对象 2、吞吐量增加 3、应用程序无意中保存了对象引用,对象无法被 GC 回收 4、应用程序过度使用 finalizer
今天看到某一篇文章的一句话 单例DCL 前面加 V 。就这句话让我把 单例模式 又仔细看了一遍。Java 中的 单例模式 是我们一直且经常使用的设计模式之一,大家都很熟悉,所以这篇文章仅仅做我自己记忆。
在任何Java面试当中多线程和并发方面的问题都是必不可少的一部分。如果你想获得更多职位,那么你应该准备很多关于多线程的问题。面试官会问面试者很多令人混淆的Java线程问题
内容以共享、参考、研究为目的,不存在任何商业目的。其版权属原作者所有,如有侵权或违规,请与小编联系!情况属实本人将予以删除!