在 Java 后端实现对 xxlJob 的任务管理

前言

最近需要在 Java 后端操作 xxl 任务调度平台来完成任务的加入，在网上找了发现全都需要修改 admin 服务，对于已经上线的项目来说，明显不好操作切繁琐，修改 admin 服务的原因也很简单，是因为在外部直接调用添加、修改任务方法时，会被拦截告诉未登陆，因此需要在源码内多补充几个方法同时加上绕过登陆的注解。那么有没有一种办法可以不修改源码也可以操作 admin 服务内的方法呢。看了 xxlJob 的源码部分发现是可行的，在页面内我们可以发现登陆 xxlJob 后会返回一个 cookie，那么我们就借助这个 cookie 来辅助我们登陆！

不了解 xxlJob 的朋友可以看一下上一篇：项目接入 xxl-job

工具类介绍及使用

我自己已经封装了一个工具类用来帮助我们操作 admin 服务，使用也很简单。首先对工具类进行一个简单的介绍，流程框架为 调用登陆方法获取 cookie >>> 携带 cookie 和参数访问指定方法 >>> 返回结果

在使用工具类之前，我们只需要手动修改三个值以及引入一个实体类即可，分别是 xxlJob 的地址、账号和密码。

实体类如下，主要是帮助我们转换为正确的调度任务对象集合

package com.sora.domain;

import lombok.Data;

import java.util.Date;

/**
 * xxl-jobInfo实体类
 */
@Data
public class XxlJobInfo {

   private int id;             // 主键ID

   private int jobGroup;     // 执行器主键ID
   private String jobDesc;

   private Date addTime;
   private Date updateTime;

   private String author;    // 负责人
   private String alarmEmail; // 报警邮件

   private String scheduleType;         // 调度类型
   private String scheduleConf;         // 调度配置，值含义取决于调度类型
   private String misfireStrategy;          // 调度过期策略

   private String executorRouteStrategy;  // 执行器路由策略
   private String executorHandler;           // 执行器，任务Handler名称
   private String executorParam;         // 执行器，任务参数
   private String executorBlockStrategy;  // 阻塞处理策略
   private int executorTimeout;          // 任务执行超时时间，单位秒
   private int executorFailRetryCount;       // 失败重试次数

   private String glueType;      // GLUE类型  #com.xxl.job.core.glue.GlueTypeEnum
   private String glueSource;    // GLUE源代码
   private String glueRemark;    // GLUE备注
   private Date glueUpdatetime;   // GLUE更新时间

   private String childJobId;    // 子任务ID，多个逗号分隔

   private int triggerStatus;    // 调度状态：0-停止，1-运行
   private long triggerLastTime;  // 上次调度时间
   private long triggerNextTime;  // 下次调度时间

}

在工具类里，我对 cookie 做了一个存入本地 map 的操作，这样只需要访问一次登陆接口，后续直接从 map 获取即可（目前没有发现 cookie 过期，如果有可以在过期后再次访问登陆接口刷新 cookie 的值）

因为 xxlJob 的参数都是 formData 格式，所以我这里添加、修改任务的时候需要把整个 map 传递过去。下面简单说一下各个方法如何使用

login（登陆）：用于获取访问 admin 服务的 cookie，不需要我们去调用，工具类会自动调用该方法并存入 cookieMap 完成 cookie 的获取

getCookie（获取 cookie）：从 cookieMap 内获取 cookie，没有 cookie 则访问 login 方法

addJob（添加调度任务）：传入一个 map，key 为 xxlJobInfo 的字段名，value 则为值，进行任务的添加

updateJob（修改调度任务）：传入一个 map，key 为 xxlJobInfo 的字段名，value 则为值，进行任务的修改

startJob（开启调度任务）：传入一个 long 类型的任务 id，将该任务的 TriggerStatus 字段设置为 1，开始执行

stopJob（停止调度任务）：传入一个 long 类型的任务 id，将该任务的 TriggerStatus 字段设置为 0，停止执行

removeJob（移除调度任务）：传入一个 long 类型的任务 id，删除该任务

pageList（获取所有调度任务）：传入执行器 id，获取该执行器下所有调度任务，返回 XxlJobInfo 类型的集合

link（访问 admin 服务）：使用 OkHttp 远程调用访问 admin，内部调用 postFormDataResponse 方法

postFormDataResponse（封装请求参数并获取响应）：封装响应体并请求接口

getHeaders（获取请求头）：获取 headers，不需要我们去调用，工具类会自动调用该方法并返回对应 headers

package com.sora.utils;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sora.domain.XxlJobInfo;
import com.sora.jackson.InitObjectMapper;
import okhttp3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.MediaType;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class XxlUtil {

    private static final Logger logger = LoggerFactory.getLogger(XxlUtil.class);

    /**
     * xxl地址
     */
    private static String xxlJobAdminAddress = "http://127.0.0.1:7777/xxl-job-admin";
    /**
     * xxl用户名
     */
    private static final String USER_NAME = "admin";
    /**
     * xxl密码
     */
    private static final String PASSWORD = "123456";
    private static final String LOGIN_URL = "/login";
    private static final String ADD_INFO_URL = "/jobinfo/add";
    private static final String REMOVE_INFO_URL = "/jobinfo/remove";
    private static final String UPDATE_INFO_URL = "/jobinfo/update";
    private static final String START_URL = "/jobinfo/start";
    private static final String STOP_URL = "/jobinfo/stop";
    private static final String PAGELIST_URL = "/jobinfo/pageList";

    private static HashMap<String, String> cookieMap = new HashMap<>();
    private static final ObjectMapper objectMapper = new ObjectMapper();

    private static OkHttpClient okHttpClient = new OkHttpClient.Builder()
            .connectTimeout(5L, TimeUnit.SECONDS)
                .readTimeout(5L, TimeUnit.SECONDS)
                .build();

    /**
     * 登陆
     */
    private static void login() {
        HashMap<String, Object> paramMap = new HashMap<>();
        paramMap.put("userName", USER_NAME);
        paramMap.put("password", PASSWORD);

        HashMap<String, String> headerMap = new HashMap<>();
        paramMap.put("Content-Type", MediaType.APPLICATION_FORM_URLENCODED.toString());

        Response response = postFormDataResponse(xxlJobAdminAddress + LOGIN_URL, paramMap, headerMap);

        // 检查响应状态码
        if (response.isSuccessful()) {
            // 提取Token
            String cookie = response.headers().get("Set-Cookie");
            cookieMap.put("cookie", cookie);
            response.close();
        } else {
            logger.error("登录失败，响应体：" + response);
        }
    }

    /**
     * 获取cookie
     *
     * @return
     */
    private static String getCookie() {
        String cookie = cookieMap.get("cookie");
        for (int i = 0; i < 3; i++) {
            if (cookie == null) {
                login();
                cookie = cookieMap.get("cookie");
            } else {
                return cookie;
            }
        }
        return null;
    }

    /**
     * 添加任务
     *
     * @param xxlJobInfoMap
     * @return
     */
    public static String addJob(HashMap<String, Object> xxlJobInfoMap) {
        return link(xxlJobAdminAddress + ADD_INFO_URL, xxlJobInfoMap);
    }

    /**
     * 修改任务信息
     *
     * @param xxlJobInfoMap
     * @return
     */
    public static String updateJob(HashMap<String, Object> xxlJobInfoMap) {
        return link(xxlJobAdminAddress + UPDATE_INFO_URL, xxlJobInfoMap);
    }

    /**
     * 启动任务
     *
     * @param jobId
     * @return
     */
    public static String startJob(long jobId) {
        HashMap<String, Object> map = new HashMap<>();
        map.put("id", String.valueOf(jobId));
        return link(xxlJobAdminAddress + START_URL, map);
    }

    /**
     * 停止任务
     *
     * @param jobId
     * @return
     */
    public static String stopJob(long jobId) {
        HashMap<String, Object> map = new HashMap<>();
        map.put("id", String.valueOf(jobId));
        return link(xxlJobAdminAddress + STOP_URL, map);
    }

    /**
     * 删除
     *
     * @param jobId
     * @return
     */
    public static String removeJob(long jobId) {
        HashMap<String, Object> map = new HashMap<>();
        map.put("id", String.valueOf(jobId));
        return link(xxlJobAdminAddress + REMOVE_INFO_URL, map);
    }

    /**
     * 查找调度任务列表
     *
     * @param
     * @return
     */
    public static List<XxlJobInfo> pageList(long group) {
        HashMap<String, Object> map = new HashMap<>();
        map.put("jobGroup", group);
        map.put("triggerStatus", -1);
        String response = link(xxlJobAdminAddress + PAGELIST_URL, map);
        try {
            String data = objectMapper.readTree(response).get("data").toString();
            List<XxlJobInfo> xxlJobInfos = objectMapper.readValue(data, new TypeReference<List<XxlJobInfo>>() {
            });
            return xxlJobInfos;
        } catch (JsonProcessingException e) {
            logger.error("json解析错误，[xxl-job工具类]调用pageList方法发生异常！！", e);
        }
        return new ArrayList<>();
    }

    /**
     * 与xxlJobAdmin进行交互
     * @param url
     * @param paramMap
     * @return
     */
    private static String link(String url, HashMap<String, Object> paramMap) {
        String cookie = getCookie();
        if (cookie == null) {
            return null;
        }
        HashMap<String, String> headerMap = new HashMap<>();
        headerMap.put("Cookie", cookie);
        try {
            Response response = postFormDataResponse(url, paramMap, headerMap);
            String body = response.body().string();
            response.close();
            return body;
        } catch (IOException e) {
            logger.error("错误请求，请求xxlJob发生异常！", e);
        }
        return null;
    }


    /**
     * 封装响应体并请求
     * @param URL 请求路径
     * @param paramMap 参数map
     * @param headerMap 请求头map
     * @return
     */
    private static Response postFormDataResponse(String URL, HashMap<String, Object> paramMap,HashMap<String,String> headerMap) {
        // 获取headers
        Headers headers = getHeaders(headerMap);

        MultipartBody.Builder data = new MultipartBody.Builder()
                .setType(MultipartBody.FORM);

        paramMap.forEach((k,v) -> {
            data.addFormDataPart(k, v.toString());
        });

        Request request = new Request.Builder()
                .post(data.build())
                .url(URL)
                .headers(headers)
                .build();

        try {
            Response execute = okHttpClient.newCall(request).execute();
            if (execute.isSuccessful()) {
                return execute;
            }
        } catch (IOException e) {
            logger.error("发起请求失败！", e);
        }
        return null;
    }

    /**
     * 根据请求头map获取对应的headers
     * @param headersMap
     * @return
     */
    private static Headers getHeaders(Map<String, String> headersMap) {
        Headers.Builder builder = new Headers.Builder();
        headersMap.forEach(builder::add);
        return builder.build();
    }
}

依赖引入

OkHttp：

<dependency>
  <groupId>com.squareup.okhttp3</groupId>
  <artifactId>okhttp</artifactId>
  <version>4.11.0</version>
</dependency>

jackson:

<dependency>
  <groupId>com.fasterxml.jackson.core</groupId>
  <artifactId>jackson-databind</artifactId>
  <version>2.15.2</version>
</dependency>
<dependency>
  <groupId>com.fasterxml.jackson.core</groupId>
  <artifactId>jackson-core</artifactId>
  <version>2.15.2</version>
</dependency>
<dependency>
  <groupId>com.fasterxml.jackson.core</groupId>
  <artifactId>jackson-annotations</artifactId>
  <version>2.15.2</version>
</dependency>

项目实践测试

新增：

这里的 key 要和 XxlJobInfo 保持一致，因为 xxlJob 需要 formData 请求，需要提交表单，所以我们需要对对象进行拆分存入 map 内，大部分的参数我都直接用值表示了，如果有疑问或者好奇，可以直接去看 XxlJobInfo 类的注释。添加成功的 content 后返回的是新增的任务 id，jobGroup 是执行器 id，可以直接去数据库的 xxl_job_group 里面查看，因为执行器我们通常不会改变，所以这里我并没有写针对执行器的方法。

@GetMapping("/jobTest")
   public Result jobTest() {
       HashMap<String, Object> paramMap = new HashMap<>();
       paramMap.put("JobGroup",2);
       paramMap.put("JobDesc","任务描述");
       paramMap.put("executorParam","我是参数");
       paramMap.put("ExecutorHandler","testHandler（handler名称）");
       paramMap.put("alarmEmail","告警邮件地址@gmail.com");
       paramMap.put("ScheduleType","CRON");
       paramMap.put("ScheduleConf","0/10 * * * * ?");
       paramMap.put("GlueType","BEAN");
       paramMap.put("MisfireStrategy","DO_NOTHING");
       paramMap.put("ExecutorBlockStrategy","SERIAL_EXECUTION");
       paramMap.put("ExecutorRouteStrategy","FIRST");
       paramMap.put("TriggerStatus",0);
       paramMap.put("Author","sora33");
       String string = XxlUtil.addJob(paramMap);
       try {
           JsonNode jsonNode = objectMapper.readTree(string);
           JsonNode content = jsonNode.get("content");
           return Result.success(content);
       } catch (JsonProcessingException e) {
           return Result.success(null);
       }
   }

查看数据库，数据添加成功

修改：

修改其实和添加一样，只不过需要我们多加入一个参数 id，传入一个已经存在的任务 id，并传入修改后的值，例如我需要把告警邮件修改了则可以如下设置（这里要把任务所有的属性都传入，哪怕是不修改的属性，不然会报缺少 xx 字段的提示）

public Result jobTest() {
  HashMap<String, Object> paramMap = new HashMap<>();
  paramMap.put("JobGroup",2);
  paramMap.put("id",47);
  paramMap.put("JobDesc","任务描述");
  paramMap.put("executorParam","我是参数");
  paramMap.put("ExecutorHandler","testHandler（handler名称）");
  paramMap.put("alarmEmail","二次告警邮件地址@gmail.com");
  paramMap.put("ScheduleType","CRON");
  paramMap.put("ScheduleConf","0/10 * * * * ?");
  paramMap.put("GlueType","BEAN");
  paramMap.put("MisfireStrategy","DO_NOTHING");
  paramMap.put("ExecutorBlockStrategy","SERIAL_EXECUTION");
  paramMap.put("ExecutorRouteStrategy","FIRST");
  paramMap.put("TriggerStatus",0);
  paramMap.put("Author","sora33");
  String string = XxlUtil.updateJob(paramMap);
}

数据库修改成功

删除 / 开启任务 / 暂停任务：

这三类任务很简单，直接传入需要的任务 id 即可。

public Result jobTest() {
    XxlUtil.startJob(47L);
    XxlUtil.stopJob(48L);
    XxlUtil.removeJob(45);
    return Result.success();
}

如下，启用 id 为 47 的调度任务，停用 48 的调度任务，删除 45 的调度任务

获取所有任务：

调用 pageList 方法，并传入执行器 id，获取该执行器下的所有任务

public Result jobTest() {
    List<XxlJobInfo> xxlJobInfos = XxlUtil.pageList(2);
    // 获取id为44的任务
    XxlJobInfo jobInfo = xxlJobInfos.stream()
            .filter(data -> data.getId() == 44)
            .findFirst()
            .orElse(null);
    return Result.success(jobInfo);
}

结束语

工具类的介绍及使用到这里就结束了，有任何想法或者疑问的可以联系我，如果本篇文章对你有用，欢迎分享给其他人，最后也到了元旦，祝各位元旦快乐～