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

Posted 2022-09-19 热爱编程的大忽悠

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

• 基于Redis实现的延迟任务
• • 编码实现
• 优缺点

---

源码链接

---

基于Redis实现的延迟任务

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

主要有以下几个问题:

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

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

这里我们先简单演示一下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.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，这里的配置方式需要进行微调。

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复用即可，所以我感觉这也算不上什么缺点吧！