Skip to content

D
datax-cloud
Commit f1857ffe by yuwei
Options

2.0.0项目初始化

parent 0c60e535
Showing with 103 additions and 22 deletions
datax-config/src/main/resources/config/datax-gateway-dev.yml
......@@ -107,4 +107,16 @@ spring:
- name: Hystrix
args:
name: dataSqlConsoleHystrix
fallbackUri: forward:/fallback
# 定时任务
- id: datax-service-quartz
uri: lb://datax-service-quartz
predicates:
- Path=/quartz/**
filters:
- SwaggerHeaderFilter
- StripPrefix=1
- name: Hystrix
args:
name: quartzHystrix
fallbackUri: forward:/fallback
\ No newline at end of file
datax-config/src/main/resources/config/datax-service-quartz-dev.yml
......@@ -2,25 +2,25 @@
spring:
redis:
database: 1
host: 192.168.234.101
host: 127.0.0.1
port: 6379
password: 1234@abcd # 密码（默认为空）
timeout: 6000ms # 连接超时时长（毫秒）
password: # 密码（默认为空）
timeout: 6000ms # 连接超时时长（毫秒）
lettuce:
pool:
max-active: 1000 # 连接池最大连接数（使用负值表示没有限制）
max-wait: -1ms # 连接池最大阻塞等待时间（使用负值表示没有限制）
max-wait: -1ms # 连接池最大阻塞等待时间（使用负值表示没有限制）
max-idle: 10 # 连接池中的最大空闲连接
min-idle: 5 # 连接池中的最小空闲连接
datasource:
mysql:
driver-class-name: com.p6spy.engine.spy.P6SpyDriver
url: jdbc:p6spy:mysql://192.168.234.100:3306/data_cloud?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8
url: jdbc:p6spy:mysql://127.0.0.1:3306/data_cloud?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8
username: root
password: 1234@abcd
quartz:
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://192.168.234.100:3306/data_cloud_quartz?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8
url: jdbc:mysql://127.0.0.1:3306/data_cloud_quartz?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8
username: root
password: 1234@abcd
type: com.zaxxer.hikari.HikariDataSource
......@@ -69,6 +69,7 @@ spring:
isClustered: true
clusterCheckinInterval: 1000
useProperties: false
misfireThreshold: 5000 # 设置临界时间为5秒
threadPool:
class: org.quartz.simpl.SimpleThreadPool
threadCount: 20
......
datax-modules/quartz-service-parent/quartz-service/src/main/java/cn/datax/service/quartz/config/StartedUpRunner.java
package cn.datax.service.quartz.config;
import cn.datax.service.quartz.api.entity.QrtzJobEntity;
import cn.datax.service.quartz.quartz.utils.ScheduleUtil;
import cn.datax.service.quartz.service.QrtzJobService;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ConfigurableApplicationContext;
......@@ -9,6 +13,7 @@ import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
@Component
@RequiredArgsConstructor
......@@ -17,6 +22,9 @@ public class StartedUpRunner implements ApplicationRunner {
private final ConfigurableApplicationContext context;
private final Environment environment;
@Autowired
private QrtzJobService qrtzJobService;
@Override
public void run(ApplicationArguments args) {
if (context.isActive()) {
......@@ -26,6 +34,10 @@ public class StartedUpRunner implements ApplicationRunner {
"端口号：" + environment.getProperty("server.port") + "\n" +
"-----------------------------------------";
System.out.println(banner);
// 项目启动时，初始化定时器
List<QrtzJobEntity> list = qrtzJobService.list();
ScheduleUtil.init(list);
}
}
}
datax-modules/quartz-service-parent/quartz-service/src/main/java/cn/datax/service/quartz/quartz/utils/ScheduleUtil.java
......@@ -2,8 +2,11 @@ package cn.datax.service.quartz.quartz.utils;
import cn.datax.common.core.DataConstant;
import cn.datax.common.exception.DataException;
import cn.datax.common.utils.ThrowableUtil;
import cn.datax.service.quartz.api.entity.QrtzJobEntity;
import cn.datax.service.quartz.quartz.ScheduleJob;
import cn.hutool.core.collection.CollUtil;
import lombok.extern.slf4j.Slf4j;
import org.quartz.CronScheduleBuilder;
import org.quartz.CronTrigger;
import org.quartz.JobBuilder;
......@@ -14,7 +17,12 @@ import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.TriggerBuilder;
import org.quartz.TriggerKey;
import org.quartz.impl.triggers.CronTriggerImpl;
import java.util.Date;
import java.util.List;
@Slf4j
public class ScheduleUtil {
public static final String QUARTZ_TASK_KEY = "__QUARTZ_TASK_KEY__";
......@@ -34,25 +42,32 @@ public class ScheduleUtil {
return JobKey.jobKey(QUARTZ_TASK_KEY + jobId);
}
/**
* 获取触发器key
*/
public static TriggerKey getTriggerKey(String jobId) {
return TriggerKey.triggerKey(QUARTZ_TASK_KEY + jobId);
}
/**
* 获取表达式触发器
*/
public static CronTrigger getCronTrigger(String jobId) {
public static CronTrigger getCronTrigger(QrtzJobEntity job) {
try {
return (CronTrigger) scheduler.getTrigger(getTriggerKey(jobId));
TriggerKey triggerKey = getTriggerKey(job.getId());
CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
// 如果不存在则创建一个定时任务
if(null == trigger){
createScheduleJob(job);
trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
}
return trigger;
} catch (SchedulerException e) {
throw new DataException("获取定时任务CronTrigger出现异常", e);
}
}
/**
* 获取触发器key
*/
public static TriggerKey getTriggerKey(String jobId) {
return TriggerKey.triggerKey(QUARTZ_TASK_KEY + jobId);
}
/**
* 创建定时任务
*/
public static void createScheduleJob(QrtzJobEntity job) {
......@@ -62,9 +77,23 @@ public class ScheduleUtil {
// 表达式调度构建器
CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(job.getCronExpression());
// 按新的cronExpression表达式构建一个新的trigger
CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity(getTriggerKey(job.getId())).withSchedule(cronScheduleBuilder).build();
CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity(getTriggerKey(job.getId()))
//withMisfireHandlingInstructionDoNothing
//——不触发立即执行
//——等待下次Cron触发频率到达时刻开始按照Cron频率依次执行
//withMisfireHandlingInstructionIgnoreMisfires
//——以错过的第一个频率时间立刻开始执行
//——重做错过的所有频率周期后
//——当下一次触发频率发生时间大于当前时间后，再按照正常的Cron频率依次执行
//withMisfireHandlingInstructionFireAndProceed
//——以当前时间为触发频率立刻触发一次执行
//——然后按照Cron频率依次执行
.withSchedule(cronScheduleBuilder.withMisfireHandlingInstructionDoNothing())
.build();
// 放入参数，运行时的方法可以获取
jobDetail.getJobDataMap().put(JOB_DATA_MAP, job);
// 重置启动时间
((CronTriggerImpl)trigger).setStartTime(new Date());
// 交给scheduler去调度
scheduler.scheduleJob(jobDetail, trigger);
// 暂停任务
......@@ -82,13 +111,17 @@ public class ScheduleUtil {
public static void updateScheduleJob(QrtzJobEntity job) {
try {
TriggerKey triggerKey = getTriggerKey(job.getId());
CronTrigger trigger = getCronTrigger(job);
// 表达式调度构建器
CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(job.getCronExpression());
CronTrigger trigger = getCronTrigger(job.getId());
// 按新的cronExpression表达式重新构建trigger
trigger = trigger.getTriggerBuilder().withIdentity(triggerKey).withSchedule(cronScheduleBuilder).build();
trigger = trigger.getTriggerBuilder().withIdentity(triggerKey)
.withSchedule(cronScheduleBuilder.withMisfireHandlingInstructionDoNothing())
.build();
// 参数
trigger.getJobDataMap().put(JOB_DATA_MAP, job);
// 重置启动时间
((CronTriggerImpl)trigger).setStartTime(new Date());
scheduler.rescheduleJob(triggerKey, trigger);
// 暂停任务
if (job.getStatus() == DataConstant.EnableState.DISABLE.getKey()) {
......@@ -139,17 +172,38 @@ public class ScheduleUtil {
*/
public static void deleteJob(String id) {
try {
TriggerKey triggerKey = getTriggerKey(id);
// 停止触发器
TriggerKey triggerKey = getTriggerKey(id);
// 停止触发器
scheduler.pauseTrigger(triggerKey);
// 移除触发器
scheduler.unscheduleJob(triggerKey);
// 暂停任务
scheduler.pauseJob(getJobKey(id));
// 删除任务
scheduler.deleteJob(getJobKey(id));
} catch (SchedulerException e) {
throw new DataException("删除定时任务失败", e);
}
}
/**
* 项目启动时，初始化定时器
*/
public static void init(List<QrtzJobEntity> list) {
if(CollUtil.isNotEmpty(list)){
for (QrtzJobEntity job : list) {
TriggerKey triggerKey = getTriggerKey(job.getId());
CronTrigger trigger = null;
try {
trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
} catch (SchedulerException e) {
log.error("全局异常信息ex={}, StackTrace={}", e.getMessage(), ThrowableUtil.getStackTrace(e));
}
// 如果不存在则创建一个定时任务
if (null == trigger) {
ScheduleUtil.createScheduleJob(job);
} else {
ScheduleUtil.updateScheduleJob(job);
}
}
}
}
}
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment