通过延时队列和mysql uodate原子性实现延迟任务

通过延时队列和mysql uodate原子性实现延迟任务

背景：公司内部通过at群聊机器人来实现一些功能，例如：@机器人 查询 user 1 这个命令就是查询用户表 id为1的操作，后端对接了机器人，由后端来解析命令 再进行相应的操作。
新需求：延迟执行命令任务，例如：@机器人 延迟执行 1s/1m/1h 查询 user 1 通过延迟执行命令来延迟执行后面的命令。

解决方案【简单版】

使用Java自带的延迟队列和mysql的update原子性来实现

• 延迟队列：延迟执行任务
  
• mysql update原子性：服务多实例部署的时候 保证只有一个实例能执行任务
  

mysql任务表

1. create table core_delay_task
2. (
3.     id                  int auto_increment comment '主键' primary key,
4.     task_param          varchar(4096) null comment '任务参数',
5.     task_status         int default 0 null comment '任务状态 0未执行 1执行中 2执行成功 3执行失败',
6.     task_execute_time   datetime      null comment '任务执行时间',
7.     task_create_time    datetime      null comment '任务创建时间',
8.     task_update_time    datetime      null comment '任务更新时间',
9.     task_execute_result varchar(2048) null comment '任务执行结果'
10. ) comment '延迟任务表';

复制代码

业务代码

1. package xxx.service.impl;
3. import com.alibaba.fastjson2.JSON;
4. import org.slf4j.Logger;
5. import org.slf4j.LoggerFactory;
6. import org.springframework.beans.factory.InitializingBean;
7. import org.springframework.beans.factory.annotation.Autowired;
8. import org.springframework.util.ObjectUtils;
10. import java.time.LocalDateTime;
11. import java.time.ZoneId;
12. import java.time.temporal.ChronoUnit;
13. import java.util.Arrays;
14. import java.util.List;
15. import java.util.concurrent.DelayQueue;
16. import java.util.concurrent.ExecutorService;
17. import java.util.concurrent.Executors;
18. import java.util.stream.Collectors;
20. @Cmd(value = {"延迟执行"}, order = 301, admin = true)
21. public class DelayTaskImpl implements CmdService, InitializingBean {
23.     @Autowired
24.     private CoreDelayTaskMapper coreDelayTaskMapper;
26.     private static final Logger LOGGER = LoggerFactory.getLogger(DelayTaskImpl.class);
28.     private final static DelayQueue<CmdServiceDelay> DELAY_QUEUE = new DelayQueue<>();
30.     @Override
31.     public RobotResp execute(RobotReq req) {
33.         ArgsUtil argsUtil = ArgsUtil.of(req);
34.         String[] args = req.getText().split("\\s+");
35.         args[1] = null;
36.         args[2] = null;
37.         String targetText = Arrays.stream(args).filter(e -> !ObjectUtils.isEmpty(e)).collect(Collectors.joining(" "));
38.         req.setText(targetText);
40.         String delay = argsUtil.index(2);
41.         String delayValue = delay.substring(0, delay.length() - 1);
42.         Long delayLongValue = 0L;
43.         if (delay.endsWith("s")) {
44.             delayLongValue = Long.parseLong(delayValue) * 1000;
45.         } else if (delay.endsWith("m")) {
46.             delayLongValue = Long.parseLong(delayValue) * 1000 * 60;
47.         } else if (delay.endsWith("h")) {
48.             delayLongValue = Long.parseLong(delayValue) * 1000 * 60 * 60;
49.         } else {
50.             return RobotResp.ok("仅支持 秒 分钟 小时");
51.         }
53.         CoreDelayTask coreDelayTask = new CoreDelayTask();
54.         coreDelayTask.setTaskParam(JSON.toJSONString(req));
55.         coreDelayTask.setTaskStatus(0);
56.         coreDelayTask.setTaskExecuteTime(LocalDateTime.now().plus(delayLongValue, ChronoUnit.MILLIS));
57.         coreDelayTask.setTaskCreateTime(LocalDateTime.now());
58.         coreDelayTask.setTaskUpdateTime(LocalDateTime.now());
60.         coreDelayTaskMapper.insertSelective(coreDelayTask);
62.         DELAY_QUEUE.add(new CmdServiceDelay(coreDelayTask.getId(), System.currentTimeMillis() + delayLongValue, req));
64.         return RobotResp.md(String.format("命令 [%s] 将在 [%s] 开始执行", targetText, coreDelayTask.getTaskExecuteTime()));
65.     }
67.     @Override
68.     public String help() {
69.         return "延迟执行 时间 命令";
70.     }
72.     @Override
73.     public void afterPropertiesSet() throws Exception {
74.         // bean初始化完毕后 查询所有等待执行的任务
75.         List<CoreDelayTask> coreDelayTasks = coreDelayTaskMapper.listCoreDelayTaskForWaitExecute();
76.         // 将等待执行的任务放入到延迟队列
77.         for (CoreDelayTask coreDelayTask : coreDelayTasks) {
78.             DELAY_QUEUE.add(new CmdServiceDelay(coreDelayTask.getId(), coreDelayTask.getTaskExecuteTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), JSON.parseObject(coreDelayTask.getTaskParam(), RobotReq.class)));
79.         }
81.         ExecutorService executor = Executors.newFixedThreadPool(1);
82.         executor.submit(() -> {
83.             while (true) {
84.                 try {
85.                     CmdServiceDelay cmdServiceDelay = DELAY_QUEUE.take();
86.                     Integer taskId = cmdServiceDelay.getTaskId();
87.                     // 利用update原子性来保证只有一个实例能获取到执行任务的锁，然后执行任务
88.                     int i = coreDelayTaskMapper.updateCoreDelayTaskForExecute(taskId, LocalDateTime.now());
89.                     if (i > 0) {
90.                         LOGGER.info("抢到锁 开始执行延迟任务 任务ID {}", taskId);
91.                         RobotReq robotReq = cmdServiceDelay.getRobotReq();
92.                         CoreDelayTask coreDelayTask = new CoreDelayTask();
93.                         coreDelayTask.setId(taskId);
95.                         try {
96.                             // 这里可以执行具体的任务
97.                             coreDelayTask.setTaskStatus(DelayTaskStatusEnum.EXECUTE_SUCCESS.getCode());
98.                         } catch (Exception e) {
99.                             coreDelayTask.setTaskStatus(DelayTaskStatusEnum.EXECUTE_FAILED.getCode());
100.                             coreDelayTask.setTaskExecuteResult(e.getMessage());
101.                         }
103.                         coreDelayTask.setTaskUpdateTime(LocalDateTime.now());
104.                        coreDelayTaskMapper.updateByPrimaryKeySelective(coreDelayTask);
105.                     } else {
106.                         LOGGER.info("未抢到锁 放弃延迟任务 {}", taskId);
107.                     }
108.                 } catch (Exception e) {
109.                     e.printStackTrace();
110.                 }
111.             }
112.         });
113.     }
114. }

复制代码

最后

上述代码基本上可以完成延迟执行任务的需求了，可以再加一点防御性代码保证代码的健壮性。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！