任务调度

1 简介

主要用于处理分布式的定时任务，由调度中心和执行器组成。调度中心和执行器两个模块可分开部署，相互分离，两者之间通过RPC进行通信，其中调度中心主要是提供一个平台，管理调度信息，发送调度请求，自己不承担业务代码，而执行器接收调度中心的调度执行业务逻辑。

2 调度中心

统一管理任务调度平台上的调度任务，负责触发调度执行，并且提供任务管理平台。

2.1 服务配置说明

AFCENTER后端服务配置文件中application-job.properties文件默认开启调度中心与执行器配置，启动后可作为调度中心与执行器使用。

## 调度中心开关
afc.job.admin.enabled=true
## 执行器开关
afc.job.executor.enabled=true

执行器端口默认9999，如启动项目出现地址占用请自行修改端口配置

## 执行器配置
#afc.job.executor.accessToken=default_token
#afc.job.executor.address=
#afc.job.executor.ip=
#afc.job.executor.port=9999
#afc.job.executor.logPath=
#afc.job.executor.logRetentionDays=

AFCENTER初始化脚本中默认提供xxl-job建表语句与初始化数据插入SQL，若应用作为调度中心时，需要在应用业务库执行xxl-job相关SQL语句。应用作为执行器时则不需要。

3 执行器

接收调度中心的调度并且执行，可以直接执行也可以集成到项目中。

3.1 服务配置说明

应用作为执行器使用时，只需要集成AFCENTER的SDK或Starter依赖，application-job.properties文件中开启执行器开关。

## 调度中心开关
afc.job.admin.enabled=false
## 执行器开关
afc.job.executor.enabled=true

4 任务调度开发与使用

若任务运行模式为BEAN模式时，需配置JobHandler，在Spring Bean实例中，开发Job方法后配置执行器执行该job，示例参考如下

import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.DataOutputStream;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;

/**
 * job开发示例（Bean模式）
 *
 * 开发步骤：
 *      1、任务开发：在Spring Bean实例中，开发Job方法；
 *      2、注解配置：为Job方法添加注解 "@XxlJob(value="自定义jobhandler名称", init = "JobHandler初始化方法", destroy = "JobHandler销毁方法")"，注解value值对应的是调度中心新建任务的JobHandler属性的值。
 *      3、执行日志：需要通过 "XxlJobHelper.log" 打印执行日志；
 *      4、任务结果：默认任务结果为 "成功" 状态，不需要主动设置；如有诉求，比如设置任务结果为失败，可以通过 "XxlJobHelper.handleFail/handleSuccess" 自主设置任务结果；
 *
 */
@Component
public class SampleXxlJobDemo {
    private static Logger logger = LoggerFactory.getLogger(SampleXxlJobDemo.class);


    /**
     * 1、简单任务示例（Bean模式）
     */
    @XxlJob("demoJobHandler1")
    public void demoJobHandler1() throws Exception {
        XxlJobHelper.log("XXL-JOB, Hello World.");

        for (int i = 0; i < 5; i++) {
            XxlJobHelper.log("beat at:" + i);
            TimeUnit.SECONDS.sleep(2);
        }
        // default success
    }


    /**
     * 2、分片广播任务
     */
    @XxlJob("shardingJobHandler1")
    public void shardingJobHandler1() throws Exception {

        // 分片参数
        int shardIndex = XxlJobHelper.getShardIndex();
        int shardTotal = XxlJobHelper.getShardTotal();

        XxlJobHelper.log("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);

        // 业务逻辑
        for (int i = 0; i < shardTotal; i++) {
            if (i == shardIndex) {
                XxlJobHelper.log("第 {} 片, 命中分片开始处理", i);
            } else {
                XxlJobHelper.log("第 {} 片, 忽略", i);
            }
        }

    }


    /**
     * 3、命令行任务
     */
    @XxlJob("commandJobHandler1")
    public void commandJobHandler1() throws Exception {
        String command = XxlJobHelper.getJobParam();
        int exitValue = -1;

        BufferedReader bufferedReader = null;
        try {
            // command process
            ProcessBuilder processBuilder = new ProcessBuilder();
            processBuilder.command(command);
            processBuilder.redirectErrorStream(true);

            Process process = processBuilder.start();
            //Process process = Runtime.getRuntime().exec(command);

            BufferedInputStream bufferedInputStream = new BufferedInputStream(process.getInputStream());
            bufferedReader = new BufferedReader(new InputStreamReader(bufferedInputStream));

            // command log
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                XxlJobHelper.log(line);
            }

            // command exit
            process.waitFor();
            exitValue = process.exitValue();
        } catch (Exception e) {
            XxlJobHelper.log(e);
        } finally {
            if (bufferedReader != null) {
                bufferedReader.close();
            }
        }

        if (exitValue == 0) {
            // default success
        } else {
            XxlJobHelper.handleFail("command exit value("+exitValue+") is failed");
        }

    }


    /**
     * 4、跨平台Http任务
     *  参数示例：
     *      "url: http://www.baidu.com\n" +
     *      "method: get\n" +
     *      "data: content\n";
     */
    @XxlJob("httpJobHandler1")
    public void httpJobHandler1() throws Exception {

        // param parse
        String param = XxlJobHelper.getJobParam();
        if (param==null || param.trim().length()==0) {
            XxlJobHelper.log("param["+ param +"] invalid.");

            XxlJobHelper.handleFail();
            return;
        }

        String[] httpParams = param.split("\n");
        String url = null;
        String method = null;
        String data = null;
        for (String httpParam: httpParams) {
            if (httpParam.startsWith("url:")) {
                url = httpParam.substring(httpParam.indexOf("url:") + 4).trim();
            }
            if (httpParam.startsWith("method:")) {
                method = httpParam.substring(httpParam.indexOf("method:") + 7).trim().toUpperCase();
            }
            if (httpParam.startsWith("data:")) {
                data = httpParam.substring(httpParam.indexOf("data:") + 5).trim();
            }
        }

        // param valid
        if (url==null || url.trim().length()==0) {
            XxlJobHelper.log("url["+ url +"] invalid.");

            XxlJobHelper.handleFail();
            return;
        }
        if (method==null || !Arrays.asList("GET", "POST").contains(method)) {
            XxlJobHelper.log("method["+ method +"] invalid.");

            XxlJobHelper.handleFail();
            return;
        }
        boolean isPostMethod = method.equals("POST");

        // request
        HttpURLConnection connection = null;
        BufferedReader bufferedReader = null;
        try {
            // connection
            URL realUrl = new URL(url);
            connection = (HttpURLConnection) realUrl.openConnection();

            // connection setting
            connection.setRequestMethod(method);
            connection.setDoOutput(isPostMethod);
            connection.setDoInput(true);
            connection.setUseCaches(false);
            connection.setReadTimeout(5 * 1000);
            connection.setConnectTimeout(3 * 1000);
            connection.setRequestProperty("connection", "Keep-Alive");
            connection.setRequestProperty("Content-Type", "application/json;charset=UTF-8");
            connection.setRequestProperty("Accept-Charset", "application/json;charset=UTF-8");

            // do connection
            connection.connect();

            // data
            if (isPostMethod && data!=null && data.trim().length()>0) {
                DataOutputStream dataOutputStream = new DataOutputStream(connection.getOutputStream());
                dataOutputStream.write(data.getBytes("UTF-8"));
                dataOutputStream.flush();
                dataOutputStream.close();
            }

            // valid StatusCode
            int statusCode = connection.getResponseCode();
            if (statusCode != 200) {
                throw new RuntimeException("Http Request StatusCode(" + statusCode + ") Invalid.");
            }

            // result
            bufferedReader = new BufferedReader(new InputStreamReader(connection.getInputStream(), "UTF-8"));
            StringBuilder result = new StringBuilder();
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                result.append(line);
            }
            String responseMsg = result.toString();

            XxlJobHelper.log(responseMsg);

            return;
        } catch (Exception e) {
            XxlJobHelper.log(e);

            XxlJobHelper.handleFail();
            return;
        } finally {
            try {
                if (bufferedReader != null) {
                    bufferedReader.close();
                }
                if (connection != null) {
                    connection.disconnect();
                }
            } catch (Exception e2) {
                XxlJobHelper.log(e2);
            }
        }

    }

    /**
     * 5、生命周期任务示例：任务初始化与销毁时，支持自定义相关逻辑；
     */
    @XxlJob(value = "demoJobHandler21", init = "init", destroy = "destroy")
    public void demoJobHandler21() throws Exception {
        XxlJobHelper.log("XXL-JOB, Hello World.");
    }
    public void init(){
        logger.info("init");
    }
    public void destroy(){
        logger.info("destroy");
    }


}

完成后端配置，启动调度中心服务与执行器服务后，打开AFCENTER平台任务调度菜单。

4.1 执行器管理

新建执行器

打开执行器管理菜单，点击新建执行器，创建模式选择应用，这里的EOS8-DEMO-LJ是集成AFCENTER的SDK依赖创建的执行器服务。选择完成后点击确定即完成执行器创建。

创建完应用执行器后，自动读取应用机器地址，可点击查看。

执行器行内操作列，可对当前行执行器进行编辑与删除等操作。

4.2 任务管理

打开任务管理菜单，点击新建任务按钮，打开任务创建弹窗，创建定时任务。执行器选择刚创建的应用执行器EOS8-DEMO-LJ，输入任务名称与报警邮件完成基本配置。

调度类型选择CRON，并输入CRON表达式，完成调度配置。

4.2.1 BEAN模式

任务配置下运行模式支持BEAN模式与逻辑流模式，此处选择BEAN模式，JobHandler中输入job的demo中使用注解@XxlJob()创建的对应任务项value值demoJobHandler1,完成任务配置。

高级配置这里遵循默认配置，点击确定完成任务创建。

创建完成后的任务可在当前行操作列点击启动，调度中心将像执行器发送调度请求，执行器接收调度中心的调度后按照调度规则执行业务逻辑。

点击操作栏编辑即可对当前任务信息进行修改。鼠标放置操作列更多操作图标即可查看到其余操作。

点击执行一次，打开执行配置弹窗，在完成配置后点击确认，系统将只调度一次任务执行,在调度日志中可查看到调度执行详情。

点击查看日志，打开当前任务执行日志列表，可通过列表中信息查看任务执行详情。

点击注册节点，打开当前任务注册节点ip地址信息查看弹窗。

点击最近执行时间，打开执行时间弹窗，可查详细执行时间。

点击删除与复制按钮可对当前任务进行删除与复制操作。

4.2.1 逻辑流

当运行模式为逻辑流时，选择需要执行的构件包下的逻辑流。本例以“定时器示例”作为执行逻辑流，示例配置调度类型为CRON，选择调度执行时间周期，点击确定。

配置完成后，点击启动任务执行。

注册节点、执行一次操作当运行模式为逻辑流时与BEAN模式一致，可参考BEAN模式配置。

点击查询日志，查询调度结果。

也可根据操作日志查询调度执行情况。

4.3 调度日志

点击调度日志菜单，打开调度日志列表，可查看任务执行的详细信息。

点击当前执行任务日志的调度备注列查看按钮，打开调度结果弹窗，可查看到任务调度配置信息与执行结果。

点击当前执行任务日志的操作列执行日志按钮，打开执行日志详情新窗口，通过日志对比该任务实现中的日志输出信息，可以看出任务已被执行并且得到我们想要的结果信息。

    /**
     * 1、简单任务示例（Bean模式）
     */
    @XxlJob("demoJobHandler1")
    public void demoJobHandler1() throws Exception {
        XxlJobHelper.log("XXL-JOB, Hello World.");

        for (int i = 0; i < 5; i++) {
            XxlJobHelper.log("beat at:" + i);
            TimeUnit.SECONDS.sleep(2);
        }
        // default success
    }

点击调度日志-清理按钮，打开日志清理配置弹窗，完成清理配置，点击确定即可对调度日志记录进行清理操作。

4.4 运行报表

点击运行报表，打开报表界面，即可查看任务执行统计等信息。

常见问题

1.启动调度任务报错： Statuscode(401) invalid.for url : http://xxx/api/afc/job/api/registry access url [/api/afc/job/api/job-registry-bizx] token [null] must be login

解决方案：需在应用user-config.xml白名单加入配置/api/afc/job/api/*，如下图所示。

2.启动任务调度失败，检查application-job.properties配置文件执行器地址是否是xxl-job的地址，参考如下：

#executor配置

#xxl-job, access token

afc.job.executor.accessToken=default_token

afc.job.executor.address=http://192.168.16.238:9999/

afc.job.executor.ip=192.168.16.238

afc.job.executor.port=9999

#其中ip和port根据实际情况配置

3.当第三方应用只作为任务执行器时，将调度配置application-job.properties关闭即为afc.job.admin.enabled=false。

当应用既是任务调度又作为执行器时，需配置xxl.job.accessToken=default_token,如下图所示。