Redis 延时任务实现及与定时任务区别详解

目录

• 引言
• 一、延时任务是什么
• 二、延时任务和定时任务的区别是什么
• 三、技术对比
• 1.数据库轮询
• 2.JDK的延迟队列
• 3.时间轮算法
• 4.使用消息队列
• 5.Redis的Zset实现延时任务
• 5.1 思路分析
• 5.2 Redis中Zset的简单介绍及使用
• 5.3 延时队列工厂
• 5.4 RedisUtil工具类
• 5.5 测试延时队列
• 四、总结

    1.add(K key, V value, double score)
    添加元素到变量中同时指定元素的分值。
    redisTemplate.opsForZSet().add("zSetValue","A",1);
    2.rangeByScore(K key, double min, double max)
    根据设置的score获取区间值。
    zSetValue = redisTemplate.opsForZSet().rangeByScore("zSetValue",1,2);
    3.rangeByScore(K key, double min, double max,long offset, long count)
    根据设置的score获取区间值从给定下标和给定长度获取最终值。
    zSetValue = redisTemplate.opsForZSet().rangeByScore("zSetValue",1,5,1,3);
    4.rangeWithScores(K key, long start, long end)
    获取RedisZSetCommands.Tuples的区间值。
    Set<ZSetOperations.TypedTuple<Object>> typedTupleSet = redisTemplate.opsForZSet().rangeWithScores("typedTupleSet",1,3);
    Iterator<ZSetOperations.TypedTuple<Object>> iterator = typedTupleSet.iterator();
    while (iterator.hasNext()){
        ZSetOperations.TypedTuple<Object> typedTuple = iterator.next();
        Object value = typedTuple.getValue();
        double score = typedTuple.getScore();
      }
    5.删除成员
    redisTemplate.opsForZSet().remove("myZset","a","b");

/**
 * 延时队列工厂
 *
 **/
@Slf4j
public abstract class AbstractDelayQueueMachineFactory {
    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private ThreadPoolTaskExecutor asyncTaskExecutor;
    /**
     * 插入任务id
     *
     * @param jobId 任务id(队列内唯一)
     * @param time  延时时间(单位 :毫秒)
     * @return 是否插入成功
     */
    public boolean addJob(String jobId, Integer time) {
        Calendar instance = Calendar.getInstance();
        //增加延时时间,获取最终触发时间
        instance.add(Calendar.MILLISECOND, time); 
        long delayMillisecond = instance.getTimeInMillis();
        log.info("延时队列添加问题{}",jobId);
        return redisUtil.zAdd(setDelayQueueName(), delayMillisecond, jobId);
    }
    /**
     * 删除任务id
     *
     * @param jobId 任务id(队列内唯一)
     */
    public boolean removeJob(String jobId) {
        Long num = redisUtil.zRemove(setDelayQueueName(), jobId);
        if (num > 0) return true;
        return false;
    }
    /**
     * 延时队列机器开始运作
     */
    private void startDelayQueueMachine() {
        log.info("延时队列{}开始启动", setDelayQueueName());
        // 监听redis队列
        while (true) {
            try {
                // 获取当前时间前的任务列表
                Set<ZSetOperations.TypedTuple<Object>> tuples = redisUtil.zRangeByScore(setDelayQueueName(), 0, System.currentTimeMillis() );
                // 如果任务不为空
                if (!CollectionUtils.isEmpty(tuples)) {
                    log.info("延时任务开始执行:{}", JSONUtil.toJsonStr(tuples));
                    Iterator<ZSetOperations.TypedTuple<Object>> iterator = tuples.iterator();
                    while (iterator.hasNext()){
                        ZSetOperations.TypedTuple<Object> typedTuple = iterator.next();
                        String questionId = Convert.toStr(typedTuple.getValue());
                        // 移除缓存，如果移除成功则表示当前线程处理了延时任务，则执行延时任务
                        // 删除成功才执行延时任务，否则不执行，这样可以避免分布式系统延时任务多次执行
                        Long num = redisUtil.zRemove(setDelayQueueName(), questionId);
                        // 如果移除成功, 则执行
                        if (num > 0) {
                            asyncTaskExecutor.execute(() -> invoke(questionId));
                        }
                    }
                }
            } catch (Exception e) {
                log.error("处理延时任务发生异常,异常原因为{}", e.getMessage(), e);
            } finally {
                // 间隔()分钟执行一次
                //根据业务场景设置对应时间
                try {
                    TimeUnit.MINUTES.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    /**
     * 最终执行的任务方法
     *
     * @param jobId 任务id
     */
    public abstract void invoke(String jobId);
    /**
     * 要实现延时队列的名字
     */
    public abstract String setDelayQueueName();
    //Spring Boot初始化时开启一条线程运行
    @PostConstruct
    public void init() {
        new Thread(this::startDelayQueueMachine).start();
    }
}

@Component
@Slf4j
public class RedisUtil {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    /**
     * 向Zset里添加成员
     *
     * @param key   key值
     * @param score 分数，通常用于排序
     * @param value 值
     * @return 增加状态
     */
    public boolean zAdd(String key, long score, String value) {
        Boolean result = redisTemplate.opsForZSet().add(key, value, score);
        return result;
    }
    /**
     * 获取 某key 下 某一分值区间的队列
     *
     * @param key  缓存key
     * @param from 开始时间
     * @param to   结束时间
     * @return 数据
     */
    public Set<ZSetOperations.TypedTuple<Object>> zRangeByScore(String key, int from, long to) {
        Set<ZSetOperations.TypedTuple<Object>> set = redisTemplate.opsForZSet().rangeByScoreWithScores(key, from, to);
        return set;
    }
    /**
     * 移除 Zset队列值
     *
     * @param key   key值
     * @param value 删除的集合
     * @return 删除数量
     */
    public Long zRemove(String key, String... value) {
        return redisTemplate.opsForZSet().remove(key, value);
    }
}

/**
 * 测试延时队列
 *
 */
@Slf4j
@Component
public class DelayQueue extends AbstractDelayQueueMachineFactory {
    @Autowired
    private ZnjExpertConsultQuestionRecordMapper questionRecordMapper;
    /** 
     * 处理业务逻辑
     */ 
    @Override
    public void invoke(String jobId) {
        Integer questionId = Convert.toInt(jobId);
        ZnjExpertConsultQuestionRecordEntity questionRecordEntity = questionRecordMapper.selectById(questionId);
        Boolean flag = znjExpertConsultService.whetherEnd(questionRecordEntity);
   /** 
    * 延时队列名统一设定
    */ 
    @Override
    public String setDelayQueueName() {
        return "expert_consult:delay_queue";
    }
}