xxl-job 动态创建一次性定时任务

需求

类似预约会议，设置提醒

1. 添加数据记录（会议开始时间、会议开始前xx分钟提醒会议人员参会）
2. 开启一个定时任务（从开始时间算，前xx分钟出发一次性任务）
3. 定时任务执行后需要删除该任务（不然会堆积大量无效任务）

特点：动态创建任务（无法硬编码设置任务@Scheduled）、任务只执行一次后丢弃

一、考虑方案

1. redis 订阅发布过期提醒
   pass: redis 服务器密码没找到 /(ㄒoㄒ)/~~
2. 定时任务每分钟查询当前时间需要执行的数据去发短信
   太耗费资源
3. xxljob 动态创建任务 任务执行完毕后销毁
   就这个吧

二、实现思路

1. gitee上下载个不要钱的源码 xxl-job
2. 本地先跑起来，换换数据库驱动，导表，修改一下数据库连接等
3. 会议服务引入xxl-job core 核心包
4. 根据xxl-job 接口地址和参数，远程调用创建任务、开始任务、删除任务等操作
5. 根据业务创建corn生成器 给xxl-job做参数

三、代码实现

3.1 引入xxl-core 核心包

<dependency><groupId>com.xuxueli</groupId><artifactId>xxl-job-core</artifactId><version>2.4.2-SNAPSHOT</version>
</dependency>

3.2 远程调用

3.2.0 yaml

xxl:job:userName: adminpassword: 123456accessToken: default_token # #调度中心通讯TOKEN [选填]：非空时启用 对应xxl-job 配置文件中的xxl.job.accessTokenadmin:addresses: http://127.0.0.1:8080/xxl-job-admin  #xxljob调度中心部署  例如：http://127.0.0.1:8080/xxl-job-adminexecutor:appname: xxl-job-pd #xxljob配置的执行器名称，ip:  #执行器IP，默认为空表示自动获取IPport: 9999  #xxljob配置的端口号，默认为9999logpath: /data/xxl-job/jobhandler  #执行器运行日志文件存储磁盘路径logretentiondays: -1  #调度中心日志表数据保存天数，过期日志自动清理；限制大于等于7时生效，否则, 如-1，关闭自动清理功能

3.2.1 配置类

package com.gsafety.bg.pd.config;import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;@Data
@Component
@ConfigurationProperties(prefix = "xxl.job")
public class XxlJobProperties {private String userName;private String password;private String accessToken;private Admin admin;private Executor executor;@Datapublic static class Admin {private String addresses;}@Datapublic static class Executor {private String appname;private String ip;private Integer port;private String logpath;private Integer logretentiondays;}
}

import com.xxl.job.core.executor.impl.XxlJobSpringExecutor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import javax.annotation.Resource;/*** @ClassName: XxlJobConfig* @Description: xxl_job配置类*/
@Configuration
public class XxlJobConfig {@Resourceprivate XxlJobProperties properties;@Beanpublic XxlJobSpringExecutor xxlJobExecutor() {XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();xxlJobSpringExecutor.setAdminAddresses(properties.getAdmin().getAddresses());xxlJobSpringExecutor.setAppname(properties.getExecutor().getAppname());xxlJobSpringExecutor.setIp(properties.getExecutor().getIp());xxlJobSpringExecutor.setPort(properties.getExecutor().getPort());xxlJobSpringExecutor.setAccessToken(properties.getAccessToken());xxlJobSpringExecutor.setLogPath(properties.getExecutor().getLogpath());xxlJobSpringExecutor.setLogRetentionDays(properties.getExecutor().getLogretentiondays());return xxlJobSpringExecutor;}
}

3.2.2 入参

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;import java.time.LocalDateTime;
import java.util.Date;/*** xxl-job info* @author xuxueli  2016-1-12 18:25:49*/
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class XxlJobReq {private String jobDesc;//任务描述private String author;// 负责人private String executorParam;// 执行器，任务参数private LocalDateTime startTime; //计划时间private long hour; //计划时间前 hour小时后执行private long min; //计划时间前 min分钟后执行
}

3.2.3 任务返回实体

import lombok.Data;@Data
public class ReturnT {private int code;private String msg;private int content;}

3.2.4 任务调用

import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gsafety.bg.gsdss.common.utils.json.JsonUtil;
import com.gsafety.bg.pd.config.XxlJobProperties;
import com.gsafety.bg.pd.model.dto.req.XxlJobReq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Configuration;import javax.annotation.Resource;
import java.io.IOException;
import java.net.HttpCookie;
import java.util.*;/*** @Description: xxljob工具类*/@Slf4j
@Configuration
public class XxlJobService {//该MAP主要用于缓存Xxl-Job的Cookieprivate static Map<String, String> loginCookie = new HashMap<>();@Resourceprivate XxlJobProperties properties;public Map<String, Object> createFormData(XxlJobReq xxlJob) {Map<String, Object> formData = new HashMap<>();formData.put("jobGroup", 2); //xxl_job_group表对应的组formData.put("jobDesc", xxlJob.getJobDesc()); //任务名称formData.put("author", xxlJob.getAuthor());  //用户姓名formData.put("alarmEmail", "");formData.put("scheduleType", "CRON");formData.put("scheduleConf", CronGenerator.cronStr(xxlJob.getStartTime(),xxlJob.getHour(),xxlJob.getMin()));   //cron表达式formData.put("schedule_conf_CRON", "");formData.put("schedule_conf_FIX_RATE", "");formData.put("schedule_conf_FIX_DELAY", "");formData.put("glueType", "BEAN");formData.put("executorHandler", "msgJobHandler");//写死了，有多个处理可以用传参方式formData.put("executorParam", xxlJob.getExecutorParam()); //传json文件的保存路径formData.put("executorRouteStrategy", "FIRST");formData.put("childJobId", "");formData.put("misfireStrategy", "DO_NOTHING");formData.put("executorBlockStrategy", "SERIAL_EXECUTION");formData.put("executorTimeout", "0");formData.put("executorFailRetryCount", "0");formData.put("glueRemark", "GLUE代码初始化");formData.put("glueSource", "");//构建好以后调用XxlJobClient的addJob方法并传递formData参数,这个类及方法的实现在下面return formData;}//添加定时任务public Integer addJob(XxlJobReq xxlJob) {//这里的url接口路径一定要是自己F12抓取到的HttpRequest request = HttpRequest.post(properties.getAdmin().getAddresses() + "/jobinfo/add").header("Content-Type", "multipart/form-data")//每次请求都需要带上Cookie,getCookie方法在后面.header("Cookie", getCookie()).form(createFormData(xxlJob));try {// 执行 HTTP POST 请求创建定时任务HttpResponse response = request.execute();String result = response.body();if (StrUtil.isNotBlank(result) && 200 == response.getStatus()) {//定时任务创建成功后拿到任务idReturnT obj = JsonUtil.of(result, ReturnT.class);log.info("定时任务创建成功，任务ID为：" + obj.toString());return obj.getContent();} else {log.error("定时任务创建失败");}} catch (Exception e) {log.info("定时任务创建失败，发生异常：" + e.getMessage());}return null;}//启动定时任务public void startJob(Integer jobId) {//创建后的定时任务默认是STOP状态,所以我们还要通过定时任务id调度任务启动接口HttpRequest requests = HttpRequest.post(properties.getAdmin().getAddresses() + "/jobinfo/start").header("Content-Type", "multipart/form-data").header("Cookie", getCookie()).form("id", jobId);//通过HTTP请求启动定时任务HttpResponse responses = requests.execute();String results = responses.body();if (StrUtil.isNotBlank(results) && 200 == responses.getStatus()) {log.info("定时任务{}启动成功。", jobId);} else {log.error("定时任务{}启动失败。", jobId);}}//停止定时任务public void stopJob(Integer jobId) {//通过定时任务id调度任务停止接口HttpRequest requests = HttpRequest.post(properties.getAdmin().getAddresses() + "/jobinfo/stop").header("Content-Type", "multipart/form-data").header("Cookie", getCookie()).form("id", jobId);//通过HTTP请求停止定时任务HttpResponse responses = requests.execute();String results = responses.body();if (StrUtil.isNotBlank(results) && 200 == responses.getStatus()) {log.info("定时任务{}停止成功。", jobId);} else {log.error("定时任务{}停止失败。", jobId);}}/*** 2.查询定时任务:* 因为是动态实时创建定时任务,所以建议单独创建一个执行器去执行这些定时任务,方便后续批量进行查询出来进行删除。* 这里入参执行器ID,通过执行器ID来查询该执行器下所有Stop状态的定时任务,因为Xxl-job定时任务默认*在执行完最后一次任务后就会自动进入STOP状态,这样查询出来所有STOP状态任务后方便我们后续进行删除清理定时任务。*/public List<Long> SelectJob(Integer jobGroup) {HttpRequest request = HttpRequest.post(properties.getAdmin().getAddresses() + "/jobinfo/pageList").header("Content-Type", "multipart/form-data").header("Cookie", getCookie()).form("jobGroup", jobGroup).form("triggerStatus", 0).form("start", 0);//执行 HTTP POST 请求启动定时任务HttpResponse response = request.execute();// 解析响应体ObjectMapper mapper = new ObjectMapper();List<Long> idList = new LinkedList<>();try {JsonNode responseNode = mapper.readTree(response.body());JsonNode dataNode = responseNode.get("data");//遍历删除id对应的定时任务if (dataNode.isArray()) {for (JsonNode node : dataNode) {Long id = node.get("id").asLong();idList.add(id);}}} catch (IOException e) {System.out.println("解析响应体时发生异常：" + e.getMessage());}return idList;}/*** 3.删除定时任务:* Xxl-job目前没有没有直接批量进行删除定时任务的,所以我们使用遍历去挨个删除,如果考虑到性能问题,单独创建一个定时来调用该删除方法即可,每天凌晨去执行该删除清理的定时任务。*/public void removalJob(List<Long> idList) {for (Long id : idList) {HttpRequest requests = HttpRequest.post(properties.getAdmin().getAddresses() + "/jobinfo/remove").header("Content-Type", "multipart/form-data").header("Cookie", getCookie()).form("id", id);//执行HTTP请求删除定时任务HttpResponse response = requests.execute();if (StrUtil.isNotBlank(response.body()) && 200 == response.getStatus()) {log.info("定时任务{}删除成功。", id);} else {log.error("定时任务{}删除失败。", id);}}}//获取Cookiepublic String getCookie() {for (int i = 0; i < 3; i++) {String cookieStr = loginCookie.get("XXL_JOB_LOGIN_IDENTITY");if (cookieStr != null) {return "XXL_JOB_LOGIN_IDENTITY=" + cookieStr;}login();}throw new RuntimeException("获取 xxl-job cookie 失败!");}//优先到MAP缓存中获取,如果没有获取到则会请求xxljob的登录来获取Cookie,这里提供三次失败可重试。public void login() {String url = properties.getAdmin().getAddresses()+"/login";HttpResponse response = HttpRequest.post(url).form("userName", properties.getUserName()).form("password", properties.getPassword()).execute();List<HttpCookie> cookies = response.getCookies();Optional<HttpCookie> cookieOpt = cookies.stream().filter(cookie -> cookie.getName().equals("XXL_JOB_LOGIN_IDENTITY")).findFirst();if (!cookieOpt.isPresent())throw new RuntimeException("获取 xxl-job cookie 失败!");String value = cookieOpt.get().getValue();loginCookie.put("XXL_JOB_LOGIN_IDENTITY", value);}}

3.3 cron生成器

import lombok.extern.slf4j.Slf4j;import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;@Slf4j
public class CronGenerator {public static String cronStr(LocalDateTime baseTime,long h,long m) {// 计算计划时间h小时m分钟后的时间点LocalDateTime delayTime = baseTime.plusHours(h).plusMinutes(m);// 创建Cron表达式String cronExpression = generateCronExpression(delayTime);log.info("Cron expression for [h] hours and [m] minutes from baseTime: " + cronExpression);return cronExpression;}public static String generateCronExpression(LocalDateTime dateTime) {// 将LocalDateTime转换为Cron表达式的格式DateTimeFormatter formatter = DateTimeFormatter.ofPattern("ss mm HH dd MM ? yyyy");// 生成Cron表达式return dateTime.format(formatter);}
}

3.4 handler实现

package com.gsafety.bg.pd.service.job;import com.xxl.job.core.context.XxlJobContext;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;import java.util.List;@Component
@Slf4j
public class JobHandlers {private XxlJobService xxlJobService;@XxlJob("msgJobHandler")public void msgJobHandler() {XxlJobHelper.log("XXL-JOB, msgJobHandler.");String jobParam = XxlJobContext.getXxlJobContext().getJobParam();log.info("任务参数: " + jobParam);XxlJobHelper.log("任务参数: " + jobParam);//解析必要参数后发送提醒，参数按业务需要传入，这里就可以接收到}@XxlJob("clearJobHandler")public void clearJobHandler() {XxlJobHelper.log("XXL-JOB, clearJobHandler.");List<Long> jobIds = xxlJobService.SelectJob(2);xxlJobService.removalJob(jobIds);XxlJobHelper.log("清理任务: " + jobIds);}
}

3.4 测试

    @PostMapping("v1/test")public String test(@RequestBody @Valid XxlJobReq xxlJob){Integer jobId = xxlJobService.addJob(xxlJob);xxlJobService.startJob(jobId);return jobId.toString();}

踩坑

1. 代码暂时还没有设置清理无效任务，可设置定时任务，每天定时执行clearJobHandler即可。
2. corn 通常为周期性任务，设置一次性任务可以直接指定年月日时间即可
   例如：CRON：00 56 15 16 07 ? 2024，即2024-07-06 15:56:00执行一次，执行后xxljob会自动将任务stop
3. 开发环境和本地调试
   xxljob bean方式执行任务，任务会自动注册到xxljob，可点击注册节点查看注册的ip，如果开发环境和本地不通，那么部署在开发环境服务器的xxljob会找不到你本地的服务，任务会执行失败。要确保注册到xxljob的ip是其可以访问到的ip。如果已经注册了访问不到的ip，可以清一下xxl_job_registry表中对应的ip记录。

---