微服务开发系列第十一篇：XXL-JOB

作者: AC编程 | 来源:发表于2023-05-18 14:26 被阅读0次

总概

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、源码地址

alanchenyan/ac-mall2-cloud

C、本节实现目标

搭建xxl-job环境
xxl-job-admin平台创建定时任务
动态创建定时任务，实现动态创建15分钟未支付自动关闭订单的定时任务

D、系列

一、部署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，则是在通过用户名和密码调用调度中心的/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;

/**
 * @author Alan Chen
 * @description 订单未支付倒计时关闭
 * @date 2023/05/17
 */
@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