延迟任务多种实现姿势--中
- 基于Redis实现的延迟任务
- 编码实现
- 优缺点
源码链接
基于Redis实现的延迟任务
如果要基于Redis来实现延迟任务,你会怎么做？
主要有以下几个问题:
- 选择什么数据结构来保存延迟任务信息
redis提供了String,List,set,hash,zset(sorted set)几种数据类型
这里我们选择采用zset数据结构来保存延迟任务的信息,zset数据结构通过score来进行排序这里我们先简单演示一下zset的基本用法:
- zset该怎么存储订单延迟任务信息
所以我们可以利用zset score这个排序的这个特性，来实现延时任务
- 在用户下单的时候，同时生成延时任务放入redis，key是可以自定义的，比如：delaytask:order
- value的值分成两个部分，一个部分是score用于排序，一个部分是member，member的值我们设置为订单对象（如：订单编号），因为后续延时任务时效达成的时候，我们需要有一些必要的订单信息（如：订单编号），才能完成订单自动取消关闭的动作。
- 延时任务实现的重点来了，score我们设置为：订单生成时间 延时时长。这样redis会对zset按照score延时时间进行排序。
- 开启redis扫描任务，获取"当前时间 > score"的延时任务并执行。即：当前时间 > 订单生成时间 延时时长的时候，执行延时任务。
编码实现
这里为了方便，采用spring-boot-starter-data-redis来快速完成对redis客户端的搭建工作。
代码语言：javascript复制<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>其次需要在Spring Boot的application.yml配置文件中，配置redis数据库的链接信息。我这里配置的是redis的单例，如果大家的生产环境是哨兵模式、或者是集群模式的redis，这里的配置方式需要进行微调。
代码语言：javascript复制spring:
redis:
database: 0 # Redis 数据库索引（默认为 0）
host: 192.168.161.3 # Redis 服务器地址
port: 6379 # Redis 服务器连接端口
password: 123456 # Redis 服务器连接密码（默认为空）
timeout: 5000 # 连接超时，单位ms
lettuce:
pool:
max-active: 8 # 连接池最大连接数（使用负值表示没有限制） 默认 8
max-wait: -1 # 连接池最大阻塞等待时间（使用负值表示没有限制） 默认 -1
max-idle: 8 # 连接池中的最大空闲连接 默认 8
min-idle: 0 # 连接池中的最小空闲连接 默认 0- redis实现的延迟队列
package com.delayTask.zset;
import com.delayTask.DelayTaskEvent;
import com.delayTask.DelayTaskQueue;
import com.delayTask.delayQueue.OrderDelayEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import javax.annotation.Resource;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* @author 大忽悠
* @create 2022/9/18 17:32
*/
@Component
@Slf4j
@RequiredArgsConstructor
public class RedisOrderDelayQueue implements DelayTaskQueue<DelayTaskEvent,DelayTaskEvent>, InitializingBean {
/**
* key
*/
private static final String ORDER_DELAY_TASK_KEY="delaytask:order";
@Resource(name = "redisTemplate")
private ZSetOperations<String,Object> zSet;
private final PollTime pollTimeManager =new PollTime();
private final ExecutorService threadPool=Executors.newFixedThreadPool(3);
/**
* <p>
* 生成一个延迟任务加入延迟队列中去
* </p>
*
* @param delayTaskEvent
* @return 可以定位此次延迟任务的标记
*/
@Override
public DelayTaskEvent produce(DelayTaskEvent delayTaskEvent) {
OrderDelayEvent orderDelayEvent = (OrderDelayEvent) delayTaskEvent;
//订单对象作为value
long delayTime = orderDelayEvent.getDelay(TimeUnit.MILLISECONDS);
//当前时间 订单延迟时间作为score
long score = System.currentTimeMillis() delayTime;
//存入redis集合中
zSet.add(ORDER_DELAY_TASK_KEY,orderDelayEvent,score);
//轮询参数调整
pollTimeManager.addDelayTask(delayTaskEvent.getDelay(TimeUnit.MILLISECONDS));
return delayTaskEvent;
}
/**
* 处理到期的延迟任务
*/
@Override
public void consume(DelayTaskEvent delayTaskEvent) {
delayTaskEvent.handleDelayEvent();
zSet.remove(ORDER_DELAY_TASK_KEY,delayTaskEvent);
}
/**
* 查询redis,看是否有延迟任务到期
*/
private void consume() {
//查询出到期时间在当前时间之前的所有任务
Set<ZSetOperations.TypedTuple<Object>> expiredTasks = zSet.rangeByScoreWithScores(ORDER_DELAY_TASK_KEY, 0, System.currentTimeMillis());
//存在到期的延迟任务
if (!CollectionUtils.isEmpty(expiredTasks)) {
//挨个处理每个过期任务
for (ZSetOperations.TypedTuple<Object> expiredTask : expiredTasks) {
consume((DelayTaskEvent) expiredTask.getValue());
//轮询参数调整
pollTimeManager.removeDelayTask();
}
}
}
/**
* <p>
* 取消taskId对应的延迟任务
* </p>
*
* @param taskId 延迟任务标记
*/
@Override
public void cancel(DelayTaskEvent taskId) {
zSet.remove(ORDER_DELAY_TASK_KEY,taskId);
((OrderDelayEvent) taskId).getOrder().submitOrder();
pollTimeManager.removeDelayTask();
}
/**
* Bean对象初始化好之后,就开始不断轮询,处理延迟任务
*/
@Override
public void afterPropertiesSet() throws Exception {
Executors.newSingleThreadExecutor().execute(()->{
while(true){
threadPool.submit(()->{
consume();
});
//睡眠指定的时间
try {
Thread.sleep(this.pollTimeManager.getPollTime());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
/**
* 计算轮询时间
*/
public static class PollTime{
private Long taskNum=0L;
private Long delayTimeSum=0L;
/**
* @return 轮询时间默认为500毫秒
*/
public Long getPollTime(){
return 500L;
}
public synchronized void addDelayTask(Long delayTime){
}
public synchronized void removeDelayTask(){
}
}
}- 测试代码
package com.dhy.redis;
import com.delayTask.DelayTaskMain;
import com.delayTask.delayQueue.OrderDelayEvent;
import com.delayTask.delayQueue.OrderDelayFactory;
import com.delayTask.zset.RedisOrderDelayQueue;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import java.util.concurrent.TimeUnit;
/**
* @author 大忽悠
* @create 2022/9/18 19:44
*/
@SpringBootTest(classes = DelayTaskMain.class)
public class RedisZSetTest {
@Autowired
private RedisOrderDelayQueue redisOrderDelayQueue;
@Test
public void testZSet() throws InterruptedException {
OrderDelayEvent orderDelay = OrderDelayFactory.newOrderDelay("大忽悠", "小风扇", 13.4, 10L);
OrderDelayEvent orderDelay1 = OrderDelayFactory.newOrderDelay("小朋友", "冰箱", 3000.0, 20L);
redisOrderDelayQueue.produce(orderDelay);
redisOrderDelayQueue.produce(orderDelay1);
Thread.sleep(TimeUnit.SECONDS.toMillis(5));
redisOrderDelayQueue.cancel(orderDelay1);
Thread.sleep(TimeUnit.SECONDS.toMillis(30));
}
}
优缺点
使用redis zset来实现延时任务的优点是：相对于本文开头介绍的两种方法，我们的延时任务是保存在redis里面的，redis具有数据持久化的机制，可以有效的避免延时任务数据的丢失。另外，redis还可以通过哨兵模式、集群模式有效的避免单点故障造成的服务中断。至于缺点嘛，我觉得没什么缺点。如果非要勉强的说一个缺点的话，那就是我们需要额外维护redis服务，增加了硬件资源的需求和运维成本。但是现在随着微服务的兴起，redis几乎已经成了应用系统的标配，redis复用即可，所以我感觉这也算不上什么缺点吧！
