1.XXL-JOB介绍
XXL-JOB是一个分布式任务调度平台，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线，开箱即用。它的有两个核心模块，一个模块叫做调度中心，另外一个模块叫做执行器，它把任务调度和任务执行分成两个部分。这样调度模块只需要负责任务的调度属性，触发调度信号。执行模块只需要接收调度信号，去执行具体的业务逻辑，两者可以各自的进行扩容和缩容。
2.管理端部署
2.1 下载源码
从github把源码工程clone到本地 https://github.com/xuxueli/xxl-job
2.2 修改配置文件
/xxl-job/xxl-job-admin/src/main/resources/application.properties
### 数据库
spring.datasource.url=jdbc:mysql://127.0.0.1:3306/xxl_job?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&serverTimezone=Asia/Shanghai
spring.datasource.username=root
spring.datasource.password=root_pwd
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
### 报警邮箱
[email protected]
[email protected]
spring.mail.password=xxx
### xxl-job, access token
xxl.job.accessToken=default_token
### 国际化，默认为 “zh_CN”/中文简体
xxl.job.i18n=zh_CN
### xxl-job日志保存天数
xxl.job.logretentiondays=30
2.3 打包部署
将xxl-job-admin模块打包部署
管理端访问地址 http://localhost:8080/xxl-job-admin，默认登录账号 “admin/123456”, 登录后运行界面如下图所示。
3.普通任务集成
3.1 添加maven依赖
<dependency>
<groupId>com.xuxueli</groupId>
<artifactId>xxl-job-core</artifactId>
<version>2.3.1</version>
</dependency>
3.2 application.yml新增配置
xxl:
job:
admin:
addresses: https://127.0.0.1/xxl-job-admin
accessToken: default_token
executor:
appname: your-executor
address:
ip:
port: 9999
logpath: /data/applogs/xxl-job/jobhandler
logretentiondays: 30
3.3 添加配置类
@Configuration
public class XxlJobConfig {
private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);
@Value("${xxl.job.admin.addresses}")
private String adminAddresses;
@Value("${xxl.job.accessToken}")
private String accessToken;
@Value("${xxl.job.executor.appname}")
private String appname;
@Value("${xxl.job.executor.address}")
private String address;
@Value("${xxl.job.executor.ip}")
private String ip;
@Value("${xxl.job.executor.port}")
private int port;
@Value("${xxl.job.executor.logpath}")
private String logPath;
@Value("${xxl.job.executor.logretentiondays}")
private int logRetentionDays;
@Bean
public XxlJobSpringExecutor xxlJobExecutor() {
logger.info(">>>>>>>>>>> xxl-job config init.");
XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
xxlJobSpringExecutor.setAppname(appname);
xxlJobSpringExecutor.setAddress(address);
xxlJobSpringExecutor.setIp(ip);
xxlJobSpringExecutor.setPort(port);
xxlJobSpringExecutor.setAccessToken(accessToken);
xxlJobSpringExecutor.setLogPath(logPath);
xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);
return xxlJobSpringExecutor;
}
}
3.4 编写执行器
常规任务
@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
}分片任务
@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.5 管理端新增执行器，新增任务
4.动态增删改启停执行的任务
大部分任务都是执行相同操作，指定时间执行，比如每天凌晨5点把所有取消的订单归档。
还有一种情况，我希望随着业务数据的改变，任务也动态的修改执行时间，不同的业务数据动态生成不同的任务，并且随时可以启动，停止，执行任务
此功能基于3.1，3.2，3.3，3.4，3.5，不需要在管理端手动添加任务
4.1 把XxlJobService封装成Dubbo接口
1.新增xxl-job-api模块，把XxlJobService移动到该模块
2.xxl-job-admin模块添加xxl-job-api,dubbo，zk的依赖
<dependency>
<groupId>com.xuxueli</groupId>
<artifactId>xxl-job-api</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo</artifactId>
<version>chd-2.7.3-v1.0.4</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-spring-boot-starter</artifactId>
<version>2.7.3</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>4.0.1</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>4.0.1</version>
</dependency>
<!--zk-->
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.13</version>
<exclusions>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
3.xxl-job-admin的application.properties添加相关配置
dubbo.application.id=xxl-job-service
dubbo.application.name=xxl-job-service
dubbo.registry.id=zk-registry
dubbo.registry.protocol=zookeeper
dubbo.registry.address=zookeeper://127.0.0.1:2181?client=curator
dubbo.registry.group=test
dubbo.registry.username=test
dubbo.registry.password=test
dubbo.scan.base-packages=com.xxl.job.admin.service
dubbo.protocol.id=dubbo
dubbo.protocol.name=dubbo
dubbo.protocol.port=36321
dubbo.provider.timeout=6000
dubbo.provider.status=threadpool
dubbo.consumer.check=false
dubbo.consumer.timeout=6000
dubbo.consumer.retries=0
4.xxl-job-admin启动类加上EnableDubbo注解
5.XxlJobServiceImpl类Service注解由spring切换成dubbo的
4.2 业务应用添加maven依赖
<dependency>
<groupId>com.xuxueli</groupId>
<artifactId>xxl-job-api</artifactId>
<version>2.3.1</version>
</dependency>
4.3 构造XxlJobInfo参数
public XxlJobInfo buildXxlJobInfo() {
XxlJobInfo xxlJobInfo = new XxlJobInfo();
xxlJobInfo.setJobGroup(JOB_GROUP);
xxlJobInfo.setScheduleType(SCHEDULE_TYPE);
xxlJobInfo.setGlueType(GLUE_TYPE);
xxlJobInfo.setExecutorRouteStrategy(EXECUTOR_ROUTE_STRATEGY);
xxlJobInfo.setMisfireStrategy(MISFIRE_STRATEGY);
xxlJobInfo.setExecutorBlockStrategy(EXECUTOR_BLOCK_STRATEGY);
xxlJobInfo.setExecutorTimeout(0);
xxlJobInfo.setExecutorFailRetryCount(0);
# 执行时间自行构造cron表达式
xxlJobInfo.setScheduleConf(scheduleConf);
return xxlJobInfo;
}4.4 使用dubbo服务对任务动态操作
@Reference(version = VersionConst.SERVICE_DEFAULT_VERSION)标签：admin,xxlJobInfo,job,private,JOB,XXL,xxlJobSpringExecutor,任务调度,xxl From： https://blog.51cto.com/u_13222507/6047030
private XxlJobService xxlJobService;
@Override
public ResponseVo insert(UserDto userDto) {
boolean ok = transactionTemplate.execute((status) -> {
ReturnT<String> reslut = new ReturnT<>(200, "success");
try {
userMapper.insert(userDto);
XxlJobInfo xxlJobInfo = buildXxlJobInfo();
reslut = xxlJobService.add(xxlJobInfo);
if (reslut.getCode() != ReturnT.SUCCESS_CODE) {
status.setRollbackOnly();
log.error("xxlJobService.add.rpc execute error,result:{}", JSON.toJSON(reslut));
return false;
}
reslut = xxlJobService.start(Integer.valueOf(reslut.getContent()));
if (reslut.getCode() != ReturnT.SUCCESS_CODE) {
status.setRollbackOnly();
log.error("xxlJobService.add.rpc execute error,result:{}", JSON.toJSON(reslut));
return false;
}
} catch (Exception e) {
status.setRollbackOnly();
log.error("user.insert.transaction execute error,user:{}", JSON.toJSON(userDto), e);
return false;
}
return true;
});
if (log.isDebugEnabled()) {
log.debug("用户新增结束，id={},result={}", userDto.getId(), ok);
}
return ok ? ResponseVo.success(userDto.getId()) : ResponseVo.fail(500, "PERSISTENT_OR_RPC_ERROR");
}