(1) 登录 CSP 平台。
(2) 在“应用管理 > 命名空间”列表页面，创建命名空间 job。
(3) 在“应用管理 > 应用组”页面，选中创建的命名空间 job，创建应用组 jobGrp。
(4) 在“应用管理 > 容器应用”页面，单击“创建”，创建应用 jobdemo1，单击“开启分布式调度”，Demo 应用示例如下：
(5) Demo 编程指导：
Xxljob 模式：兼容 xxljob 的 bean 模式，使用@Xxljob 注解，开发定时任务，注解的值即后续“创建任务”中的“JobHandler”参数。
示例代码：
package com.example.xxljobexecutor.job;
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 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;
@Slf4j
@Component
public class SampleXxlJob {
/**
* 1、简单任务示例（Bean 模式）
*/
@XxlJob("demoJobHandler")
public void demoJobHandler() 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("shardingJobHandler")
public void shardingJobHandler() 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("commandJobHandler")
public void commandJobHandler() 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("httpJobHandler")
public void httpJobHandler() throws Exception {
// param parse
String param = XxlJobHelper.getJobParam();
if (param==null || param.trim().length()==0) {
XxlJobHelper.log("param["+ param +"] invalid.");
XxlJobHelper.handleFail();
}
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 = "demoJobHandler2", init = "init", destroy = "destroy")
public void demoJobHandler2() throws Exception {
XxlJobHelper.log("XXL-JOB, Hello World.");
}
public void init(){
log.info("init");
}
public void destroy(){
log.info("destory");
}
}
Shell、Python、PHP、Nodejs 等脚本语言：利用执行器所在机器的环境执行脚本，即在 GLUE 代码框中，自己输入代码。前提是执行器所在的机器有对应的环境，如果是容器应用，使用镜像部署，自己准备镜像，确保镜像中有对应的环境；如果是 ECS 应用，请在 ECS 中安装相应的环境。
Go 语言代码：
package main
import (
"fmt"
xxl "github.com/xxl-job/xxl-job-executor-go"
"github.com/xxl-job/xxl-job-executor-go/example/task"
"log"
)
func main() {
exec := xxl.NewExecutor(
xxl.ServerAddr("http://127.0.0.1/xxl-job-admin"),
xxl.AccessToken(""), //请求令牌(默认为空)
xxl.ExecutorIp("127.0.0.1"), //可自动获取
xxl.ExecutorPort("9999"), //默认 9999（非必填）
xxl.RegistryKey("golang-jobs"), //执行器名称
xxl.SetLogger(&logger{}), //自定义日志
)
exec.Init()
//设置日志查看 handler
exec.LogHandler(func(req *xxl.LogReq) *xxl.LogRes {
return &xxl.LogRes{Code: xxl.SuccessCode, Msg: "", Content: xxl.LogResContent{
FromLineNum: req.FromLineNum,
ToLineNum: 2,
LogContent: "这个是自定义日志 handler",
IsEnd: true,
}}
})
//注册任务 handler
exec.RegTask("task.test", task.Test)
exec.RegTask("task.test2", task.Test2)
exec.RegTask("task.panic", task.Panic)
log.Fatal(exec.Run())
}
//xxl.Logger 接口实现
type logger struct{}
func (l *logger) Info(format string, a ...interface{}) {
fmt.Println(fmt.Sprintf("自定义日志 - "+format, a...))
}
func (l *logger) Error(format string, a ...interface{}) {
log.Println(fmt.Sprintf("自定义日志 - "+format, a...))
}
(6) Demo 配置文件配置项说明：
xxl.job.admin.addresses：xxljob 调度中心地址，默认值即可，通过 CSP 部署容器应用时会被替换为 CSP 下的 xxljob 调度中心地址。
xxl.job.accessToken：访问调度中心携带的 token，默认为空即可。
xxl.job.executor.appname：执行器名称，默认值即可，通过 CSP 部署容器应用时会和容器应用名保持一致。
xxl.job.executor.address：执行器地址，建议为空，为空时会自动获取本机的 IP 地址，通过 CSP 部署容器应用时会被替换为可调度的地址。
xxl.job.executor.ip：执行器 IP，默认为空即可。
xxl.job.executor.port：执行器端口，默认 9999 即可，通过 CSP 部署容器应用时会被替换，暂时固定为 9999。
xxl.job.executor.logpath：执行器日志路径，默认值即可。
xxl.job.executor.logretentiondays：日志过期时间，默认值即可。
(7) 在“应用管理 > 容器应用”页面，创建开启了分布式调度开关的应用后，会默认在“分布式调度 > 应用管理”页面创建同名执行器应用。
(8) 使用 DEMO 生成的 jar 包，部署容器应用。暴露两个端口，其中 8082 为业务端口，9999 为执行器内置服务器端口，用于和调度中心通信使用。
(9) 部署成功后，DEMO 应用同时作为执行器会自动注册到调度中心。
(10) 在“分布式调度 > 任务管理”页面，选中刚才创建的执行器应用，创建任务。
(11) 执行任务：执行一次或根据“调度类型”下的调度规则启动定时任务。
(12) 查看日志。
