在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);
}

结束语

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