延迟任务多种实现姿势--中

---

源码链接

---

基于Redis实现的延迟任务

如果要基于Redis来实现延迟任务,你会怎么做？

主要有以下几个问题:

1. 选择什么数据结构来保存延迟任务信息

redis提供了String,List,set,hash,zset(sorted set)几种数据类型
这里我们选择采用zset数据结构来保存延迟任务的信息,zset数据结构通过score来进行排序
1
2

这里我们先简单演示一下zset的基本用法:

---

2. zset该怎么存储订单延迟任务信息

所以我们可以利用zset score这个排序的这个特性，来实现延时任务

• 在用户下单的时候，同时生成延时任务放入redis，key是可以自定义的，比如：delaytask:order
• value的值分成两个部分，一个部分是score用于排序，一个部分是member，member的值我们设置为订单对象（如：订单编号），因为后续延时任务时效达成的时候，我们需要有一些必要的订单信息（如：订单编号），才能完成订单自动取消关闭的动作。
• 延时任务实现的重点来了，score我们设置为：订单生成时间 +延时时长。这样redis会对zset按照score延时时间进行排序。
• 开启redis扫描任务，获取"当前时间 > score"的延时任务并执行。即：当前时间 > 订单生成时间 + 延时时长的时候，执行延时任务。

---

编码实现

这里为了方便，采用spring-boot-starter-data-redis来快速完成对redis客户端的搭建工作。

<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-data-redisartifactId>
dependency>
<dependency>
    <groupId>org.apache.commonsgroupId>
    <artifactId>commons-pool2artifactId>
dependency>
1
2
3
4
5
6
7
8

其次需要在Spring Boot的application.yml配置文件中，配置redis数据库的链接信息。我这里配置的是redis的单例，如果大家的生产环境是哨兵模式、或者是集群模式的redis，这里的配置方式需要进行微调。

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
1
2
3
4
5
6
7
8
9
10
11
12
13

---

• 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);

    /**
     * 

     * 生成一个延迟任务加入延迟队列中去
     * 
     *
     * @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();
            }
        }
    }

    /**
     * 

     * 取消taskId对应的延迟任务
     * 
     *
     * @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(){

        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

• 测试代码

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));
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

---

优缺点

使用redis zset来实现延时任务的优点是：相对于本文开头介绍的两种方法，我们的延时任务是保存在redis里面的，redis具有数据持久化的机制，可以有效的避免延时任务数据的丢失。另外，redis还可以通过哨兵模式、集群模式有效的避免单点故障造成的服务中断。至于缺点嘛，我觉得没什么缺点。如果非要勉强的说一个缺点的话，那就是我们需要额外维护redis服务，增加了硬件资源的需求和运维成本。但是现在随着微服务的兴起，redis几乎已经成了应用系统的标配，redis复用即可，所以我感觉这也算不上什么缺点吧！