延迟队列服务

@SpringBootApplication
public class ScheduleApplication {
    public static void main(String[] args) {
        SpringApplication.run(ScheduleApplication.class, args);
    }
}

public class CodeGenerate {
    public static void OnMac() {
        // 代码生成器
        AutoGenerator mpg = new AutoGenerator();

        // 全局配置
        GlobalConfig gc = new GlobalConfig();
        String projectPath = System.getProperty("user.dir");
        gc.setAuthor("sw-code");
        gc.setOpen(false); // 是否打开文件资源管理器
        gc.setFileOverride(true); // 是否覆盖
        gc.setServiceName("%sService"); // 去Service的I前缀
        gc.setSwagger2(false); // 实体属性 Swagger2 注解
        gc.setIdType(IdType.ASSIGN_ID); // 主键策略
        gc.setDateType(DateType.ONLY_DATE); // 定义生成的实体类中日期类型
        mpg.setGlobalConfig(gc);

        // 数据源配置
        DataSourceConfig dsc = new DataSourceConfig();
        dsc.setUrl("jdbc:mysql:///leadnews_schedule?useSSL=false&serverTimezone=UTC&characterEncoding=utf-8&nullCatalogMeansCurrent=true");
        // dsc.setSchemaName("public");
        dsc.setDriverName("com.mysql.cj.jdbc.Driver");
        dsc.setUsername("root");
        dsc.setPassword("swx852345");
        mpg.setDataSource(dsc);


        /*
         * 包配置
         * 简单来讲 就是写绝对路径
         */
        PackageConfig pc = new PackageConfig();
        //  pc.setModuleName("code");
        pc.setParent("com.swx");
        //指定生成文件的绝对路径
        Map<String, String> pathInfo  = new HashMap<>();
        String packageName = "schedule";
        String parentPath = "/src/main/java/com/swx";
        String otherPath ="/leadnews-service/leadnews-schedule/src/main/java/com/swx/" + packageName;

        String entityPackageName = "schedule";
        pc.setEntity("model." + entityPackageName + ".pojo");
        pc.setMapper(packageName + ".mapper");
        pc.setService(packageName + ".service");
        pc.setServiceImpl(packageName + ".service.impl");
        pc.setController(packageName + ".controller.v1");

        String entityPath = projectPath.concat("/leadnews-model").concat(parentPath).concat("/model/" + entityPackageName + "/pojo");
        String mapper_path = projectPath.concat(otherPath).concat("/mapper");
        String mapper_xml_path = projectPath.concat("/leadnews-service/leadnews-schedule").concat("/src/main/resources/mapper");
        String service_path = projectPath.concat(otherPath).concat("/service");
        String service_impl_path = projectPath.concat(otherPath).concat("/service/impl");
        String controller_path = projectPath.concat(otherPath).concat("/controller/v1");

        pathInfo.put("entity_path",entityPath);
        pathInfo.put("mapper_path",mapper_path);
        pathInfo.put("xml_path",mapper_xml_path);
        pathInfo.put("service_path",service_path);
        pathInfo.put("service_impl_path",service_impl_path);
        pathInfo.put("controller_path",controller_path);
        pc.setPathInfo(pathInfo);
        mpg.setPackageInfo(pc);

        // 策略配置
        StrategyConfig strategy = new StrategyConfig();
        strategy.setInclude("taskinfo", "taskinfo_logs");
        strategy.setNaming(NamingStrategy.underline_to_camel);
        strategy.setColumnNaming(NamingStrategy.underline_to_camel);
        strategy.setEntityLombokModel(true);
        // 字段注解
        strategy.setEntityTableFieldAnnotationEnable(true);
        strategy.setRestControllerStyle(true);
        strategy.setControllerMappingHyphenStyle(true);
        mpg.setStrategy(strategy);

        mpg.execute();
    }

    public static void main(String[] args) {
        OnMac();
    }
}

/**
 * 参数
 */
@TableField("parameters")
private byte[] parameters;

@Data
@EqualsAndHashCode(callSuper = false)
@TableName("taskinfo_logs")
public class TaskinfoLogs implements Serializable {
    ....;
    
    /**
     * 版本号,用乐观锁
     */
    @Version
    @TableField("version")
    private Integer version;
    
    ....;
}

@Configuration
@MapperScan("com.swx.schedule.mapper")
public class MyBatisPlusConfig {

    /**
     * 乐观所支持
     */
    @Bean
    public MybatisPlusInterceptor optimisticLockerInnerInterceptor() {
        MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
        interceptor.addInnerInterceptor(new OptimisticLockerInnerInterceptor());
        return interceptor;
    }
}

public enum TaskLogStatus {
    SCHEDULED(0, "初始化状态"),
    EXECUTED(1, "已执行状态"),
    CANCELLED(2, "已取消状态");

    private int status;
    private String desc;

    TaskLogStatus(int status, String desc) {
        this.status = status;
        this.desc = desc;
    }

    public int status() {
        return status;
    }

    public String desc() {
        return desc;
    }
}

public class ScheduleConstants {

    public static final String REDIS_KEY_FUTURE = "future_"; // 延迟任务的Redis前缀
    public static final String REDIS_KEY_TOPIC = "topic_"; // 立即执行的Redis前缀
    public static final String FUTURE_TASK_SYNC = "FUTURE_TASK_SYNC";
    public static final Integer DEFAULT_TIME_MINUTE = 5; // 预设任务时间
    public static final Integer LOCK_TIME_MILLISECONDS = 1000 * 30;
}

public interface TaskService {

    /**
     * 添加延迟任务
     *
     * @param task 任务
     * @return 任务ID
     */
    public long addTask(Task task);
}

@Slf4j
@Service
public class TaskServiceImpl implements TaskService {

    private final TaskinfoMapper taskinfoMapper;
    private final TaskinfoLogsMapper taskinfoLogsMapper;
    private final CacheService cacheService;

    public TaskServiceImpl(TaskinfoMapper taskinfoMapper, TaskinfoLogsMapper taskinfoLogsMapper, CacheService cacheService) {
        this.taskinfoMapper = taskinfoMapper;
        this.taskinfoLogsMapper = taskinfoLogsMapper;
        this.cacheService = cacheService;
    }


    /**
     * 添加延迟任务
     *
     * @param task 任务
     * @return 任务ID
     */
    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public long addTask(Task task) {
        // 添加任务到数据库中
        boolean success = addTaskToDb(task);
        if (!success) {
            throw new RuntimeException("事务添加到数据库失败");
        }
        // 添加任务到Redis
        addTaskToRedis(task);

        return task.getTaskId();
    }

    /**
     * 把任务添加到Redis中
     *
     * @param task
     */
    private void addTaskToRedis(Task task) {
        String key = task.getTaskType() + "_" + task.getPriority();
        // 获取5分钟之后的时间
        Calendar instance = Calendar.getInstance();
        instance.add(Calendar.MINUTE, ScheduleConstants.DEFAULT_TIME_MINUTE);
        long nextScheduleTime = instance.getTimeInMillis();

        // 执行时间小于等于当前时间，即立即执行
        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            cacheService.lLeftPush(ScheduleConstants.REDIS_KEY_TOPIC + key, JSON.toJSONString(task));
        } else if (task.getExecuteTime() <= nextScheduleTime) {
            // 如果任务的执行时间大于当前时间 && 小于预设时间（未来5分钟），存入zset中
            cacheService.zAdd(ScheduleConstants.REDIS_KEY_FUTURE + key, JSON.toJSONString(task), task.getExecuteTime());
        }
    }

    /**
     * 添加任务到数据库
     *
     * @param task 任务
     * @return 是否添加成功
     */
    private boolean addTaskToDb(Task task) {
        // 保存任务表
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(task, taskinfo);
        taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
        int insert = taskinfoMapper.insert(taskinfo);
        // 回填taskID
        task.setTaskId(taskinfo.getTaskId());

        // 保存日志数据
        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(taskinfo, taskinfoLogs);
        taskinfoLogs.setVersion(1);
        taskinfoLogs.setStatus(TaskLogStatus.SCHEDULED.status());
        int insert1 = taskinfoLogsMapper.insert(taskinfoLogs);

        return insert > 0 && insert1 > 0;
    }
}

/**
 * 取消任务
 *
 * @param taskId 任务ID
 * @return 是否取消成功
 */
public boolean cancelTask(long taskId);

/**
 * 取消任务
 *
 * @param taskId 任务ID
 * @return 是否取消成功
 */
@Override
@Transactional(rollbackFor = Exception.class)
public boolean cancelTask(long taskId) {
    // 删除任务更新日志
    Task task = updateDb(taskId, TaskLogStatus.CANCELLED.status());

    if (task == null) return false;
    // 删除redis数据
    removeTaskFromCache(task);
    return true;
}

/**
 * 删除任务，更新任务日志
 *
 * @param taskId 任务ID
 * @param status 状态
 */
private Task updateDb(long taskId, int status) {
    // 删除任务
    taskinfoMapper.deleteById(taskId);

    // 更新任务日志
    TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskId);
    if (taskinfoLogs == null) {
        return null;
    }
    taskinfoLogs.setStatus(status);
    taskinfoLogsMapper.updateById(taskinfoLogs);

    Task task = new Task();
    BeanUtils.copyProperties(taskinfoLogs, task);
    task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());

    return task;
}

/**
 * 删除Redis中的数据
 *
 * @param task 任务
 */
private void removeTaskFromCache(Task task) {
    String key = task.getTaskType() + "_" + task.getPriority();
    // 执行时间小于等于当前时间，即立即执行
    if (task.getExecuteTime() <= System.currentTimeMillis()) {
        cacheService.lRemove(ScheduleConstants.REDIS_KEY_TOPIC + key, 0, JSON.toJSONString(task));
    } else {
        cacheService.zRemove(ScheduleConstants.REDIS_KEY_FUTURE + key);
    }
}

/**
 * 按照类型和优先级拉取任务
 *
 * @param type     类型
 * @param priority 优先级
 * @return 任务
 */
public Task poll(int type, int priority);

/**
 * 按照类型和优先级拉取任务
 *
 * @param type     类型
 * @param priority 优先级
 * @return 任务
 */
@Override
@Transactional(rollbackFor = Exception.class)
public Task poll(int type, int priority) {
    // 从Redis中拉取数据

    String key = type + "_" + priority;
    String taskJson = cacheService.lRightPop(ScheduleConstants.REDIS_KEY_TOPIC + key);
    if (!StringUtils.hasText(taskJson)) {
        return null;
    }
    Task task = JSON.parseObject(taskJson, Task.class);
    // 修改数据库信息
    updateDb(task.getTaskId(), TaskLogStatus.EXECUTED.status());
    return task;
}

/**
 * 未来数据定时刷新，每分钟一次
 */
@Scheduled(cron = "0 */1 * * * ?")
public void refresh() {
    String token = cacheService.tryLock(ScheduleConstants.FUTURE_TASK_SYNC, ScheduleConstants.LOCK_TIME_MILLISECONDS);
    if (StringUtils.hasText(token)) {
        log.info("未来数据定时刷新--定时任务");
        // 获取所有未来数据的集合key
        Set<String> futureKeys = cacheService.scan(ScheduleConstants.REDIS_KEY_FUTURE + "*");
        for (String futureKey : futureKeys) {
            // 获取当前数据的key
            String topicKey = ScheduleConstants.REDIS_KEY_TOPIC + futureKey.split(ScheduleConstants.REDIS_KEY_FUTURE)[1];
            // 按照key和分值查询符合条件的数据
            Set<String> tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());
            if (!tasks.isEmpty()) {
                cacheService.refreshWithPipeline(futureKey, topicKey, tasks);
                log.info("成功的将" + futureKey + "刷新到了" + topicKey);
            }
        }
    }
}

/**
 * 加锁
 *
 * @param name   key
 * @param expire 过期时间
 */
public String tryLock(String name, long expire) {
    name = name + "_lock";
    String token = UUID.randomUUID().toString();
    RedisConnectionFactory factory = stringRedisTemplate.getConnectionFactory();
    RedisConnection connection = factory.getConnection();
    try {
        // set key value [Ex seconds] [PX milliseconds] [NX/XX]
        Boolean result = connection.set(
                name.getBytes(),
                token.getBytes(),
                Expiration.from(expire, TimeUnit.MILLISECONDS),
                RedisStringCommands.SetOption.SET_IF_ABSENT // NX
        );
        if (result != null && result) {
            return token;
        }
    } finally {
        RedisConnectionUtils.releaseConnection(connection, factory, false);
    }
    return null;
}

/**
 * 未来数据定时刷新，每分钟一次
 */
@Scheduled(cron = "0 */1 * * * ?")
public void refresh() {
    String token = cacheService.tryLock(ScheduleConstants.FUTURE_TASK_SYNC, ScheduleConstants.LOCK_TIME_MILLISECONDS);
    if (StringUtils.hasText(token)) {
        log.info("未来数据定时刷新--定时任务");
        // 获取所有未来数据的集合key
        Set<String> futureKeys = cacheService.scan(ScheduleConstants.REDIS_KEY_FUTURE + "*");
        for (String futureKey : futureKeys) {
            // 获取当前数据的key
            String topicKey = ScheduleConstants.REDIS_KEY_TOPIC + futureKey.split(ScheduleConstants.REDIS_KEY_FUTURE)[1];
            // 按照key和分值查询符合条件的数据
            Set<String> tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());
            if (!tasks.isEmpty()) {
                cacheService.refreshWithPipeline(futureKey, topicKey, tasks);
                log.info("成功的将" + futureKey + "刷新到了" + topicKey);
            }
        }
    }
}

public static final String FUTURE_TASK_SYNC = "FUTURE_TASK_SYNC;
public static final Integer LOCK_TIME_MILLISECONDS = 1000 * 30;

/**
 * 数据库任务定时同步到Redis
 */
@PostConstruct
@Scheduled(cron = "0 */5 * * * ?")
public void reloadData() {
    // 清除缓存中的数据
    clearCache();
    // 查询符合条件的任务，小于未来5分钟的数据
    Calendar instance = Calendar.getInstance();
    instance.add(Calendar.MINUTE, ScheduleConstants.DEFAULT_TIME_MINUTE);
    List<Taskinfo> taskinfos = taskinfoMapper.selectList(Wrappers.<Taskinfo>lambdaQuery().lt(Taskinfo::getExecuteTime, instance.getTime()));
    // 把任务添加到redis
    if (taskinfos!=null&& !taskinfos.isEmpty()) {
        for (Taskinfo taskinfo : taskinfos) {
            Task task = new Task();
            BeanUtils.copyProperties(taskinfo, task);
            task.setExecuteTime(taskinfo.getExecuteTime().getTime());
            addTaskToRedis(task);
        }
    }

    log.info("数据库的任务同步到了Redis");
}

/**
 * 清理缓存中的数据
 */
public void clearCache() {
    Set<String> topicKeys = cacheService.scan(ScheduleConstants.REDIS_KEY_TOPIC + "*");
    Set<String> futureKeys = cacheService.scan(ScheduleConstants.REDIS_KEY_FUTURE + "*");
    cacheService.delete(topicKeys);
    cacheService.delete(futureKeys);
}
}

@FeignClient("leadnews-schedule")
public interface IScheduleClient {

    /**
     * 添加延迟任务
     *
     * @param task 任务
     * @return 任务ID
     */
    @PostMapping("/api/v1/task/add")
    public R addTask(@RequestBody Task task);

    /**
     * 取消任务
     *
     * @param taskId 任务ID
     * @return 是否取消成功
     */
    @GetMapping("/api/v1/task/{taskId}")
    public R cancelTask(@PathVariable("taskId") long taskId);

    /**
     * 按照类型和优先级拉取任务
     *
     * @param type     类型
     * @param priority 优先级
     * @return 任务
     */
    @GetMapping("/api/v1/task/{type}/{priority}")
    public R poll(@PathVariable("type") int type, @PathVariable("priority") int priority);
}

@RestController
public class ScheduleClient implements IScheduleClient {

    private final TaskService taskService;

    public ScheduleClient(TaskService taskService) {
        this.taskService = taskService;
    }

    /**
     * 添加延迟任务
     *
     * @param task 任务
     * @return 任务ID
     */
    @Override
    @PostMapping("/api/v1/task/add")
    public R addTask(@RequestBody Task task) {
        return R.success(taskService.addTask(task));
    }

    /**
     * 取消任务
     *
     * @param taskId 任务ID
     * @return 是否取消成功
     */
    @Override
    @GetMapping("/api/v1/task/{taskId}")
    public R cancelTask(@PathVariable("taskId") long taskId) {
        return R.success(taskService.cancelTask(taskId));
    }

    /**
     * 按照类型和优先级拉取任务
     *
     * @param type     类型
     * @param priority 优先级
     * @return 任务
     */
    @Override
    @GetMapping("/api/v1/task/{type}/{priority}")
    public R poll(@PathVariable("type") int type, @PathVariable("priority") int priority) {
        return R.success(taskService.poll(type, priority));
    }
}

public interface WmNewsTaskService {

    /**
     * 添加任务到延迟队列中
     *
     * @param id          文章ID
     * @param publishTime 发布时间，即任务执行时间
     */
    public void addNewsToTask(Integer id, Date publishTime);
}

@Slf4j
@Service
public class WmNewsTaskServiceImpl implements WmNewsTaskService {

    private final IScheduleClient scheduleClient;

    public WmNewsTaskServiceImpl(IScheduleClient scheduleClient) {
        this.scheduleClient = scheduleClient;
    }

    /**
     * 添加任务到延迟队列中
     *
     * @param id          文章ID
     * @param publishTime 发布时间，即任务执行时间
     */
    @Override
    @Async
    public void addNewsToTask(Integer id, Date publishTime) {

        log.info("添加任务到延迟服务中----begin");

        Task task = new Task();
        task.setExecuteTime(publishTime.getTime());
        task.setTaskType(TaskTypeEnum.NEWS_SCAN_TIME.taskType());
        task.setPriority(TaskTypeEnum.NEWS_SCAN_TIME.priority());
        WmNews wmNews = new WmNews();
        wmNews.setId(id);
        task.setParameters(ProtostuffUtil.serialize(wmNews));
        scheduleClient.addTask(task);

        log.info("添加任务到延迟服务中----end");
    }
}

@Override
@Transactional(rollbackFor = Exception.class)
public void submitNews(WmNewsDTO dto) {
    ....;
    // 异步调用自动审核
    // wmNewsAutoScanService.autoScanWmNews(wmNews.getId());
    // 添加到任务队列中
    wmNewsTaskService.addNewsToTask(wmNews.getId(), wmNews.getPublishTime());
}

/**
 * 消费任务
 */
public void scanNewsByTask();

/**
 * 消费任务，每秒拉取一次
 */
@Override
@Scheduled(fixedRate = 1000)
public void scanNewsByTask() {

    log.info("消费任务，审核文章");

    R result = scheduleClient.poll(TaskTypeEnum.NEWS_SCAN_TIME.taskType(), TaskTypeEnum.NEWS_SCAN_TIME.priority());
    if (result == null) {
        log.error("拉取任务失败");
        return;
    }
    if (result.getCode().equals(ResultCodeEnum.SUCCESS.code()) && result.getData() != null) {
        Task task = JSON.parseObject(JSON.toJSONString(result.getData()), Task.class);
        WmNews wmNews = ProtostuffUtil.deserialize(task.getParameters(), WmNews.class);
        wmNewsAutoScanService.autoScanWmNews(wmNews.getId());
    }
}

@SpringBootApplication
@EnableDiscoveryClient
@MapperScan("com.swx.wemedia.mapper")
@EnableFeignClients(basePackages = "com.swx.apis")
@EnableAsync // 开启异步任务
@EnableScheduling // 开启调度任务
public class WemediaApplication {
    public static void main(String[] args) {
        SpringApplication.run(WemediaApplication.class, args);
    }
}