总概
A、技术栈
- 开发语言：Java 1.8
- 数据库：MySQL、Redis、MongoDB、Elasticsearch
- 微服务框架：Spring Cloud Alibaba
- 微服务网关：Spring Cloud Gateway
- 服务注册和配置中心：Nacos
- 分布式事务：Seata
- 链路追踪框架：Sleuth
- 服务降级与熔断：Sentinel
- ORM框架：MyBatis-Plus
- 分布式任务调度平台：XXL-JOB
- 消息中间件：RocketMQ
- 分布式锁：Redisson
- 权限：OAuth2
- DevOps：Jenkins、Docker、K8S
B、本节实现目标
- 搭建xxl-job环境
- xxl-job-admin平台创建定时任务
- 动态创建定时任务，实现动态创建15分钟未支付自动关闭订单的定时任务
一、部署xxl-job
1.1 下载xxl-job源码
下载地址：https://github.com/xuxueli/xxl-job
xxl-job源码
1.2 初始化“调度数据库”
调度数据库初始化SQL脚本” 位置为：/xxl-job/doc/db/tables_xxl_job.sql。
调度中心支持集群部署，集群情况下各节点务必连接同一个MySQL实例。如果MySQL做主从，调度中心集群节点务必强制走主库。
1.3 部署调度中心 xxl-job-admin
1.3.1 xxl-job-admin项目
xxl-job源码里有3个项目：xxl-job-admin、xxl-job-core、xxl-job-executor-samples，
- xxl-job-admin：调度中心
- xxl-job-core：公共依赖
- xxl-job-executor-samples：执行器Sample示例（选择合适的版本执行器，可直接使用，也可以参考其并将现有项目改造成执行器）
我们部署调度中心需要用到xxl-job-admin
1.3.2 启动xxl-job-admin
将xxl-job-admin调度中心项目用IDEA打开，配置文件地址：/xxl-job/xxl-job-admin/src/main/resources/application.properties
我们修改三个地方：
- 端口：端口默认是8080，这里我们将端口改成8081
- 数据库配置：修改数据库地址和账号密码
- accessToken：默认是default_token
修改完后启动xxl-job-admin
修改端口
修改数据库地址和账号密码
修改accessToken
1.3.3 访问xxl-job-admin
调度中心访问地址：http://localhost:8081/xxl-job-admin ，该地址执行器将会使用到，作为回调地址。
默认登录账号 “admin/123456”, 登录后运行界面如下图所示。
xxl-job-admin
二、配置部署“执行器项目”
执行器项目：xxl-job-executor-sample-springboot，可直接使用，也可以参考其并将现有项目改造成执行器。
执行器项目作用：负责接收“调度中心”的调度并执行；可直接部署执行器，也可以将执行器集成到现有业务项目中。
我们不用xxl-job-executor-sample-springboot，而是直接用mall-order项目来做为执行器项目。
2.1 新增执行管理器
先在xxl-job-admin上新增一个执行管理器：executor-order
新增执行管理器
新增执行管理器
2.2 maven依赖
在mall-pom项目的pom.xml里引入xxl-job-core的maven依赖
<dependency>
<groupId>com.xuxueli</groupId>
<artifactId>xxl-job-core</artifactId>
<version>${xxljob.version}</version>
</dependency>
xxl-job-core
2.3 执行器组件配置
执行器组件，配置文件地址：/xxl-job/xxl-job-executor-samples/xxl-job-executor-sample-springboot/src/main/java/com/xxl/job/executor/core/config/XxlJobConfig.java
将该代码复制到mall-order项目
XxlJobConfig
2.4 配置xxl-job连接信息
修改mall-order服务里的bootstrap-dev.yml配置信息
server:
port: 7030
spring:
application:
name: mall-order
cloud:
nacos:
config:
server-addr: 127.0.0.1:8848
namespace: dev_id
file-extension: yml
shared-configs:
- data-id: common.yml
group: DEFAULT_GROUP
refresh: true
discovery:
namespace: dev_id
swagger:
enabled: true
title: 订单服务
basePackage: com.ac.order.controller
version: 1.0
description: 订单服务相关接口
xxl:
job:
# 执行器通讯TOKEN [选填]：非空时启用；
accessToken: 123456
admin:
# 调度中心部署跟地址 [选填]：如调度中心集群部署存在多个地址则用逗号分隔。执行器将会使用该地址进行"执行器心跳注册"和"任务结果回调"；为空则关闭自动注册；
addresses: http://127.0.0.1:8081/xxl-job-admin
executor:
# 执行器AppName [选填]：执行器心跳注册分组依据；为空则关闭自动注册
app-name: executor-order
# 执行器注册 [选填]：优先使用该配置作为注册地址，为空时使用内嵌服务 ”IP:PORT“ 作为注册地址。从而更灵活的支持容器类型执行器动态IP和动态映射端口问题。
address: ''
# 执行器IP [选填]：默认为空表示自动获取IP，多网卡时可手动设置指定IP，该IP不会绑定Host仅作为通讯实用；地址信息用于 "执行器注册" 和 "调度中心请求并触发任务"；
ip: ''
# 执行器端口号 [选填]：小于等于0则自动获取；默认端口为9999，单机部署多个执行器时，注意要配置不同执行器端口；
port: -1
# 执行器运行日志文件存储磁盘路径 [选填] ：需要对该路径拥有读写权限；为空则使用默认路径；
log-path: /data/logs/task-log
# 执行器日志保存天数 [选填] ：值大于3时生效，启用执行器Log文件定期清理功能，否则不生效；
log-retention-days: -1
修改配置
2.5 启动mall-order服务
启动mall-order服务，mall-order服务会自动注册到executor-order执行器下，此时，mall-order就是一个执行器项目。
executor-order执行器
三、xxl-job-admin平台创建定时任务
BEAN模式（方法形式)，xxl-job-admin平台创建定时任务
3.1 新建定时任务执行方法类
/**
* @author Alan Chen
* @description xxl-job-admin平台创建定时任务
* @date 2023/05/17
*/
@Slf4j
@Component
public class TaskByAdminCreateJob {
@XxlJob(value = XXLJobHandlerConstant.TASK_BY_ADMIN_CREATE)
public void doJob() {
try {
// 获取任务ID
long jobId = XxlJobHelper.getJobId();
log.info("TaskByTimeJob doJob,jobId={},param={}", jobId, XxlJobHelper.getJobParam());
// 获取任务参数
String[] params = StrUtil.splitToArray(XxlJobHelper.getJobParam(), ',');
if (params.length <= 1) {
String error = StrUtil.format("TaskByTimeJob.doJob,失败, 原因: 参数缺失, 任务ID: {}", jobId);
log.info(error);
XxlJobHelper.handleFail(error);
return;
}
// 业务逻辑
String memberId = params[0];
String memberName = params[1];
String logInfo = StrUtil.format("TaskByTimeJob.doJob,成功,memberId={},memberName={}", memberId, memberName);
log.info(logInfo);
XxlJobHelper.handleSuccess(logInfo);
} catch (Exception e) {
String error = StrUtil.format("TaskByTimeJob.doJob,失败, msg={}", e.getMessage());
log.info(error);
XxlJobHelper.handleFail(error);
}
}
}
package com.ac.common.constant;
public class XXLJobHandlerConstant {
/**
* xxl-job-admin平台创建定时任务
*/
public static final String TASK_BY_ADMIN_CREATE = "TASK_BY_ADMIN_CREATE";
}
定时执行任务
3.2 新建任务管理
新建任务管理
设置执行时间
配置JobHandler，和TaskByAdminCreateJob里配置的@XxlJob保持一致
配置
任务新建完后，需要手动启动
启动
3.3 执行任务
启动状态下的任务，可以立即执行一次
执行一次
执行参数
后台打印结果
四、代码动态创建定时任务
4.1 背景说明
xxl-job-admin平台手动创建定时任务，使用起来虽然方便，可以有时候，我们就是需要在代码中动态创建一个定时任务，而不是到页面上进行配置。比如用户下单后，我们需要动态创建一个15分钟未支付自动关闭订单的定时任务。
4.2 xxljob接口梳理
我们先到github上拉一份xxl-job的源码下来，结合着文档和代码，先梳理一下各个模块都是干什么的：
-
xxl-job-admin：任务调度中心，启动后就可以访问管理页面，进行执行器和任务的注册、以及任务调用等功能了
-
xxl-job-core：公共依赖，项目中使用到xxl-job时要引入的依赖包
-
xxl-job-executor-samples：执行示例，分别包含了springboot版本和不使用框架的版本
为了弄清楚注册和查询executor和jobHandler调用的是哪些接口，我们先从页面上去抓一个请求看看：
执行管理器接口
好了，这样就能定位到xxl-job-admin模块中xxl-job-admin/jobgroup/pageList这个接口。
按照这个思路，可以找到下面这几个关键接口：
/jobgroup/pageList：执行器列表的条件查询
/jobgroup/save：添加执行器
/jobinfo/pageList：任务列表的条件查询
/jobinfo/add：添加任务
但是如果直接调用这些接口，那么就会发现它会跳转到xxl-job-admin的的登录页面。
其实想想也明白，出于安全性考虑，调度中心的接口也不可能允许裸调的。那么再回头看一下刚才页面上的请求就会发现，它在Headers中添加了一条名为XXL_JOB_LOGIN_IDENTITY的cookie：
cookie
至于这条cookie，则是在通过用户名和密码调用调度中心的/login接口时返回的，在返回的response可以直接拿到。只要保存下来，并在之后每次请求时携带，就能够正常访问其他接口了。
到这里，我们需要的5个接口就基本准备齐了，接下来准备开始正式的改造工作。
4.3 动态创建定时任务实现
4.3.1 XxlJobInfo和XxlJobGroup类
在调用调度中心的接口前，先把xxl-job-admin模块中的XxlJobInfo和XxlJobGroup这两个类拿到我们的mall-common项目中，用于接收接口调用的结果。
XxlJobGroup
4.3.2 登录接口
在调用业务接口前，需要通过登录接口获取cookie，并在获取到cookie后，缓存到本地的Map中。
private final Map<String, String> loginCookie = new HashMap<>();
private final String adminAddresses = "http://127.0.0.1:8081/xxl-job-admin";
private final String username = "admin";
private final String password = "123456";
public String login() {
String url = adminAddresses + "/login";
HttpResponse response = HttpRequest.post(url)
.form("userName", username)
.form("password", password)
.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("get xxl-job cookie error!");
String value = cookieOpt.get().getValue();
loginCookie.put("XXL_JOB_LOGIN_IDENTITY", value);
log.info("XxlJobComponent.login.token={}", value);
return value;
}
4.3.3 获取cookie
其他接口在调用时，直接从缓存中获取cookie，如果缓存中不存在则调用/login接口，为了避免这一过程失败，允许最多重试3次。
public 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("get xxl-job cookie error!");
}
4.3.4 通过appName获取执行管理器ID
/**
* 通过appName获取执行管理器ID
*
* @param appName
* @return
*/
private int getJobGroupId(String appName) {
List<XxlJobGroup> jobGroupList = listJobGroup(appName);
if (CollectionUtil.isEmpty(jobGroupList)) {
return -1;
}
return jobGroupList.get(0).getId();
}
/**
* 获取执行管理器列表
*
* @param appName
* @return
*/
private List<XxlJobGroup> listJobGroup(String appName) {
String url = adminAddresses + "/jobgroup/pageList";
HttpResponse response = HttpRequest.post(url)
.form("appname", appName)
.cookie(getCookie())
.execute();
String body = response.body();
JSONArray array = JSONUtil.parse(body).getByPath("data", JSONArray.class);
List<XxlJobGroup> list = array.stream()
.map(o -> JSONUtil.toBean((JSONObject) o, XxlJobGroup.class))
.collect(Collectors.toList());
return list;
}
4.3.5 创建&启动定时任务
/**
* 启动定时任务
*
* @param jobId
* @return
*/
private boolean startJob(Integer jobId) {
String url = adminAddresses + "/jobinfo/start";
Map<String, Object> paramMap = new HashMap<>();
paramMap.put("id", jobId);
HttpResponse response = HttpRequest.post(url)
.form(paramMap)
.cookie(getCookie())
.execute();
JSON json = JSONUtil.parse(response.body());
Object code = json.getByPath("code");
return code.equals(200);
}
/**
* 创建定时任务
*
* @param xxlJobInfo
* @return
*/
private Integer addJobInfo(XxlJobInfo xxlJobInfo) {
String url = adminAddresses + "/jobinfo/add";
Map<String, Object> paramMap = BeanUtil.beanToMap(xxlJobInfo);
HttpResponse response = HttpRequest.post(url)
.form(paramMap)
.cookie(getCookie())
.execute();
JSON json = JSONUtil.parse(response.body());
Object code = json.getByPath("code");
if (code.equals(200)) {
Object content = json.getByPath("content");
if (content == null) {
return -1;
}
return Integer.valueOf((String) content);
}
log.info("创建定时任务失败");
return -1;
}
4.3.5 XxlJobComponent完整代码
package com.ac.order.component;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.json.JSON;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.ac.common.xxljob.XxlJobGroup;
import com.ac.common.xxljob.XxlJobInfo;
import com.ac.order.cmd.AddDefaultXxlJobCmd;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.net.HttpCookie;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
@Slf4j
@Component
public class XxlJobComponent {
private final Map<String, String> loginCookie = new HashMap<>();
private final String adminAddresses = "http://127.0.0.1:8081/xxl-job-admin";
private final String username = "admin";
private final String password = "123456";
/**
* 创建定时定时任务并启动
*
* @param cmd
* @return
*/
public boolean addAndStartJob(AddDefaultXxlJobCmd cmd) {
int jobGroup = getJobGroupId(cmd.getAppName());
if (jobGroup == -1) {
log.error("获取执行管理器ID失败,appName={}", cmd.getAppName());
return false;
}
cmd.setJobGroup(jobGroup);
XxlJobInfo jobInfo = convertDefaultJobInfo(cmd);
//创建定时任务
Integer id = this.addJobInfo(jobInfo);
if (id == -1) {
log.error("创建定时任务失败,cmd={}", cmd);
return false;
}
//启动定时任务
return this.startJob(id);
}
/**
* xxl-job登录
*
* @return
*/
public String login() {
String url = adminAddresses + "/login";
HttpResponse response = HttpRequest.post(url)
.form("userName", username)
.form("password", password)
.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("get xxl-job cookie error!");
String value = cookieOpt.get().getValue();
loginCookie.put("XXL_JOB_LOGIN_IDENTITY", value);
log.info("XxlJobComponent.login.token={}", value);
return value;
}
/**
* 其他接口在调用时，直接从缓存中获取cookie，如果缓存中不存在则调用/login接口，为了避免这一过程失败，允许最多重试3次
*
* @return
*/
public 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("get xxl-job cookie error!");
}
/**
* 通过appName获取执行管理器ID
*
* @param appName
* @return
*/
private int getJobGroupId(String appName) {
List<XxlJobGroup> jobGroupList = listJobGroup(appName);
if (CollectionUtil.isEmpty(jobGroupList)) {
return -1;
}
return jobGroupList.get(0).getId();
}
/**
* 获取执行管理器列表
*
* @param appName
* @return
*/
private List<XxlJobGroup> listJobGroup(String appName) {
String url = adminAddresses + "/jobgroup/pageList";
HttpResponse response = HttpRequest.post(url)
.form("appname", appName)
.cookie(getCookie())
.execute();
String body = response.body();
JSONArray array = JSONUtil.parse(body).getByPath("data", JSONArray.class);
List<XxlJobGroup> list = array.stream()
.map(o -> JSONUtil.toBean((JSONObject) o, XxlJobGroup.class))
.collect(Collectors.toList());
return list;
}
/**
* 启动定时任务
*
* @param jobId
* @return
*/
private boolean startJob(Integer jobId) {
String url = adminAddresses + "/jobinfo/start";
Map<String, Object> paramMap = new HashMap<>();
paramMap.put("id", jobId);
HttpResponse response = HttpRequest.post(url)
.form(paramMap)
.cookie(getCookie())
.execute();
JSON json = JSONUtil.parse(response.body());
Object code = json.getByPath("code");
return code.equals(200);
}
/**
* 创建定时任务
*
* @param xxlJobInfo
* @return
*/
private Integer addJobInfo(XxlJobInfo xxlJobInfo) {
String url = adminAddresses + "/jobinfo/add";
Map<String, Object> paramMap = BeanUtil.beanToMap(xxlJobInfo);
HttpResponse response = HttpRequest.post(url)
.form(paramMap)
.cookie(getCookie())
.execute();
JSON json = JSONUtil.parse(response.body());
Object code = json.getByPath("code");
if (code.equals(200)) {
Object content = json.getByPath("content");
if (content == null) {
return -1;
}
return Integer.valueOf((String) content);
}
log.info("创建定时任务失败");
return -1;
}
/**
* 定时任务对象转换
*
* @param cmd
* @return
*/
private XxlJobInfo convertDefaultJobInfo(AddDefaultXxlJobCmd cmd) {
XxlJobInfo jobInfo = new XxlJobInfo();
/*基础配置*/
jobInfo.setJobGroup(cmd.getJobGroup());
jobInfo.setJobDesc(cmd.getJobDesc());
jobInfo.setAuthor("SYSTEM");
jobInfo.setAlarmEmail("test.126.com");
//调度配置
jobInfo.setScheduleType("CRON");
jobInfo.setScheduleConf(cmd.getScheduleConf());
//任务配置
jobInfo.setGlueType("BEAN");
jobInfo.setExecutorHandler(cmd.getExecutorHandler());
jobInfo.setExecutorParam(cmd.getExecutorParam());
/*高级配置*/
//路由策略
jobInfo.setExecutorRouteStrategy("CONSISTENT_HASH");
//调度过期策略 DO_NOTHING忽略 FIRE_ONCE_NOW立即执行一次
jobInfo.setMisfireStrategy("FIRE_ONCE_NOW");
//阻塞处理策略
jobInfo.setExecutorBlockStrategy("SERIAL_EXECUTION");
return jobInfo;
}
}
4.4 订单未付款自动关闭15分钟倒计时
4.4.1 订单任务类
package com.ac.order.task;
import cn.hutool.core.util.StrUtil;
import com.ac.common.constant.XXLJobHandlerConstant;
import com.ac.core.util.DateUtil;
import com.ac.order.cmd.AddDefaultXxlJobCmd;
import com.ac.order.component.XxlJobComponent;
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 javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
/**
*
* @description 订单未支付倒计时关闭
*
*/
@Slf4j
@Component
public class AutoCancelOrderJob {
@Resource
private XxlJobComponent xxlJobComponent;
/**
* 订单未付款自动关闭15分钟倒计时
*
* @param orderNo
*/
public void addJob(String orderNo) {
try {
log.info("AutoCancelOrderJob.addJob,orderNo={}", orderNo);
String executorParam = orderNo;
LocalDateTime now = LocalDateTime.now();
//6小时后执行
LocalDateTime offset = DateUtil.offset(now, 15, ChronoUnit.MINUTES);
String scheduleConf = DateUtil.getCron(cn.hutool.core.date.DateUtil.date(offset));
AddDefaultXxlJobCmd cmd = AddDefaultXxlJobCmd.builder()
.appName("executor-order")
.jobDesc("订单未付款自动关闭15分钟倒计时")
.scheduleConf(scheduleConf)
.executorHandler(XXLJobHandlerConstant.AUTO_CANCEL_ORDER)
.executorParam(executorParam)
.build();
xxlJobComponent.addAndStartJob(cmd);
} catch (Exception e) {
log.info("AutoCancelOrderJob.addJob,启动任务失败,msg={}", e.getMessage());
}
}
@XxlJob(value = XXLJobHandlerConstant.AUTO_CANCEL_ORDER)
public void doJob() {
try {
// 获取任务ID
long jobId = XxlJobHelper.getJobId();
log.info("AutoCancelOrderJob.doJob,jobId={},param={}", jobId, XxlJobHelper.getJobParam());
// 获取任务参数
String orderNo = XxlJobHelper.getJobParam();
// 业务逻辑
log.info("模拟业务逻辑,AutoCancelOrderJob.doJob,关闭订单,orderNo={}", orderNo);
String logInfo = StrUtil.format("AutoCancelOrderJob.doJob,成功关闭订单,orderNo={}", orderNo);
log.info(logInfo);
XxlJobHelper.handleSuccess(logInfo);
} catch (Exception e) {
String error = StrUtil.format("AutoCancelOrderJob.doJob,失败, msg={}", e.getMessage());
log.info(error);
XxlJobHelper.handleFail(error);
}
}
}
4.4.2 Controller
@ApiOperation(value = "订单未付款自动关闭15分钟倒计时")
@GetMapping("autoCancelOrder")
public boolean autoCancelOrder(@RequestParam String orderNo) {
autoCancelOrderJob.addJob(orderNo);
return true;
}
4.4.3 测试
Postman
xxljob-admin
执行一次
参数自动填充
控制台日志
