Redis实现定时任务_代码007(未授权)

Redi s 定时任务的核心在于”@Sch e d ul e“注解，Redi s Zset，Li s t 数据结构，Redi s 管道技术

1.前端用户发起定时任务创建定时任务任务，像定时任务模块发起定时任务请求并且携带必要参数

public long addTask(Taskinfo taskinfo) {
    if (null == taskinfo.getExecuteTime()){
        //保存消息到数据库中
        Calendar executeTime = Calendar.getInstance();
        executeTime.add(Calendar.MINUTE, 5);
        Date futureTime = executeTime.getTime();
        taskinfo.setExecuteTime(futureTime);
    }
    taskInfoMapper.insert(taskinfo);
    //判断当前时间与执行时间的关系
    if (new Date().compareTo(taskinfo.getExecuteTime()) &gt; 0){
        String key =  taskinfo.getTaskType() + "_" + taskinfo.getPriority();
        /**
         * 执行时间超过当前日期直接加入到未来任务执行队列当中
         */
        cacheService.zAdd(ScheduleConstants.FUTURE + key, JSON.toJSONString(taskinfo),taskinfo.getExecuteTime().getTime());
    }
    TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
    BeanUtils.copyProperties(taskinfo, taskinfoLogs);
    taskinfoLogs.setVersion(1);
    taskinfoLogs.setStatus(ScheduleConstants.TaskStatus.INIT);
    //保存到日志表中
    taskInfoLogsService.save(taskinfoLogs);
    return taskinfo.getTaskId();
}

如果小于当前系统时间（任务补偿机制。这种情况有可能出现在于由于微服务架构中一般定时任务被单独抽取成为一个模块，模块中通信可能基于消息队列，对于消息阻塞的情况可能会出现定时任务时间小于当前时间），直接存储到Redis的Zset 集合中，并设置Score为当前时间戳(只占用4个字节时间从 1970-01-01 到 2060年)，否则则存入数据库中。

2.数据库表结构，对于定时任务我设计了两张表，一张表为taskInfo,另外一张为taskInfoLog表。日志表的作用在于记录任务是否完成，用于排查异常并且对未执行的任务进行补偿机制。

/**
 * 五分钟执行一次从数据库中得到数据存到redis中
 */
@Scheduled(cron = "0 0/5 * * * ?")
public void schedule(){
  	//尝试获取分布式锁
    String lock = cacheService.tryLock(ScheduleConstants.LOCK_NAME, 3000);
    if (Strings.isBlank(lock)){
        log.info("当前主机没有抢占到redis分布式锁，故不执行刷新定时任务");
        return;
    }
    log.info("-----------从数据库中查询任务存入redis中-----------");
    Calendar future = Calendar.getInstance();
    future.add(Calendar.MINUTE, 5);
    LambdaQueryWrapper<Taskinfo> taskinfoLambdaQueryWrapper = new LambdaQueryWrapper<>();
    //时间区间为：[当前时间，当前时间+5分钟]
    LambdaQueryWrapper<Taskinfo> queryWrapper = taskinfoLambdaQueryWrapper.le(Taskinfo::getExecuteTime, future.getTime())
            .ge(Taskinfo::getExecuteTime, new Date());
    List<Taskinfo> taskinfos = taskInfoMapper.selectList(queryWrapper);
    if (!taskinfos.isEmpty()){
        //找出了任务
        //存入redis的zset集合中去
        log.info("从DB中查询出来的需要放入redis中未来队列的数据有：{}",taskinfos);
        for (Taskinfo taskinfo : taskinfos) {
            //存入zset集合中去
            String key = taskinfo.getTaskType() + "_" + taskinfo.getPriority();
            cacheService.zAdd(ScheduleConstants.FUTURE + key,JSON.toJSONString(taskinfo),System.currentTimeMillis());
        }
    }
    log.info("-----------从数据库中查询任务存入redis中结束-----------");
}

@Scheduled(cron = "0 0/1 * * * ? ")
public void refreshRedisTask(){
    log.info("---------------刷新future队列中的任务开始------------");
    //模糊查询所有未来任务队列
    Set<String> likeKey = cacheService.scan(ScheduleConstants.FUTURE + "*");
    likeKey.forEach(key ->{
        //从futureKey中得到所有
        Set<String> taskKeys = cacheService.zRange(key, System.currentTimeMillis(), System.currentTimeMillis() + 1 * 60 * 1000);
        if (taskKeys.isEmpty()){
            log.info("当前未来任务队列中没有任务");
            return;
        }
        String consumerKey = ScheduleConstants.CONSUMER + key.substring(key.indexOf("_") + 1);
        String futureKey = ScheduleConstants.FUTURE + key.substring(key.indexOf("_") + 1);
        log.info("取出{}条任务转存到消费队列中去：{}", taskKeys.size(),taskKeys);
        cacheService.refreshWithPipeline(futureKey,consumerKey,taskKeys);
    });
}