mirror of
http://git.pig4cloud.com/pig/pigx.git
synced 2026-11-11 05:19:00 +08:00
feat(quartz): 增强多租户定时任务处理逻辑
- 在getJobKey和getTriggerKey方法中增加TenantContextHolder的租户ID兜底逻辑 - 当SysJob中tenantId为空时自动使用当前线程上下文租户ID - 优化方法注释使其更符合JavaDoc规范
This commit is contained in:
1 parent
d16d8bc18a
commit
dacdb83989
1 file changed
+208
-193
+208
-193
@@ -17,6 +17,7 @@
|
||||
|
||||
package com.pig4cloud.pigx.daemon.quartz.util;
|
||||
|
||||
import com.pig4cloud.pigx.common.data.tenant.TenantContextHolder;
|
||||
import com.pig4cloud.pigx.daemon.quartz.config.PigxQuartzFactory;
|
||||
import com.pig4cloud.pigx.daemon.quartz.constants.PigxQuartzEnum;
|
||||
import com.pig4cloud.pigx.daemon.quartz.entity.SysJob;
|
||||
@@ -24,6 +25,8 @@ import lombok.extern.slf4j.Slf4j;
|
||||
import org.quartz.*;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.Objects;
|
||||
|
||||
/**
|
||||
* 定时任务的工具类
|
||||
*
|
||||
@@ -33,212 +36,224 @@ import org.springframework.stereotype.Component;
|
||||
@Component
|
||||
public class TaskUtil {
|
||||
|
||||
/**
|
||||
* 获取定时任务的唯一key
|
||||
* @param sysjob
|
||||
* @return
|
||||
*/
|
||||
public static JobKey getJobKey(SysJob sysjob) {
|
||||
return JobKey.jobKey(sysjob.getJobName(), sysjob.getJobGroup());
|
||||
}
|
||||
/**
|
||||
* 获取定时任务的唯一key
|
||||
*
|
||||
* @param sysJob 定时任务信息
|
||||
* @return 定时任务的唯一标识
|
||||
*/
|
||||
public static JobKey getJobKey(SysJob sysjob) {
|
||||
Long tenantId = sysjob.getTenantId();
|
||||
if (Objects.isNull(tenantId)) {
|
||||
tenantId = TenantContextHolder.getTenantId();
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取定时任务触发器cron的唯一key
|
||||
* @param sysjob
|
||||
* @return
|
||||
*/
|
||||
public static TriggerKey getTriggerKey(SysJob sysjob) {
|
||||
return TriggerKey.triggerKey(sysjob.getJobName(), sysjob.getJobGroup());
|
||||
}
|
||||
// 在jobGroup中包含租户ID以实现多租户隔离
|
||||
String jobGroupWithTenant = sysjob.getJobGroup() + "_tenant_" + tenantId;
|
||||
return JobKey.jobKey(sysjob.getJobName(), jobGroupWithTenant);
|
||||
}
|
||||
|
||||
/**
|
||||
* 添加或更新定时任务
|
||||
* @param sysjob
|
||||
* @param scheduler
|
||||
*/
|
||||
public void addOrUpateJob(SysJob sysjob, Scheduler scheduler) {
|
||||
CronTrigger trigger = null;
|
||||
try {
|
||||
JobKey jobKey = getJobKey(sysjob);
|
||||
// 获得触发器
|
||||
TriggerKey triggerKey = getTriggerKey(sysjob);
|
||||
trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
|
||||
// 判断触发器是否存在(如果存在说明之前运行过但是在当前被禁用了,如果不存在说明一次都没运行过)
|
||||
if (trigger == null) {
|
||||
// 新建一个工作任务 指定任务类型为串接进行的
|
||||
JobDetail jobDetail = JobBuilder.newJob(PigxQuartzFactory.class).withIdentity(jobKey).build();
|
||||
// 将任务信息添加到任务信息中
|
||||
jobDetail.getJobDataMap().put(PigxQuartzEnum.SCHEDULE_JOB_KEY.getType(), sysjob);
|
||||
// 将cron表达式进行转换
|
||||
CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(sysjob.getCronExpression());
|
||||
cronScheduleBuilder = this.handleCronScheduleMisfirePolicy(sysjob, cronScheduleBuilder);
|
||||
// 创建触发器并将cron表达式对象给塞入
|
||||
trigger = TriggerBuilder.newTrigger()
|
||||
.withIdentity(triggerKey)
|
||||
.withSchedule(cronScheduleBuilder)
|
||||
.build();
|
||||
// 在调度器中将触发器和任务进行组合
|
||||
scheduler.scheduleJob(jobDetail, trigger);
|
||||
}
|
||||
else {
|
||||
CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(sysjob.getCronExpression());
|
||||
cronScheduleBuilder = this.handleCronScheduleMisfirePolicy(sysjob, cronScheduleBuilder);
|
||||
// 按照新的规则进行
|
||||
trigger = trigger.getTriggerBuilder()
|
||||
.withIdentity(triggerKey)
|
||||
.withSchedule(cronScheduleBuilder)
|
||||
.build();
|
||||
// 将任务信息更新到任务信息中
|
||||
trigger.getJobDataMap().put(PigxQuartzEnum.SCHEDULE_JOB_KEY.getType(), sysjob);
|
||||
// 重启
|
||||
scheduler.rescheduleJob(triggerKey, trigger);
|
||||
}
|
||||
// 如任务状态为暂停
|
||||
if (sysjob.getJobStatus().equals(PigxQuartzEnum.JOB_STATUS_NOT_RUNNING.getType())) {
|
||||
this.pauseJob(sysjob, scheduler);
|
||||
}
|
||||
}
|
||||
catch (SchedulerException e) {
|
||||
log.error("添加或更新定时任务,失败信息:{}", e.getMessage());
|
||||
}
|
||||
}
|
||||
/**
|
||||
* 获取定时任务触发器的唯一键
|
||||
*
|
||||
* @param sysJob 定时任务信息
|
||||
* @return 触发器键,包含任务名称和带租户ID的任务组
|
||||
*/
|
||||
public static TriggerKey getTriggerKey(SysJob sysjob) {
|
||||
Long tenantId = sysjob.getTenantId();
|
||||
if (Objects.isNull(tenantId)) {
|
||||
tenantId = TenantContextHolder.getTenantId();
|
||||
}
|
||||
|
||||
/**
|
||||
* 立即执行一次任务
|
||||
*/
|
||||
public static boolean runOnce(Scheduler scheduler, SysJob sysJob) {
|
||||
try {
|
||||
// 参数
|
||||
JobDataMap dataMap = new JobDataMap();
|
||||
dataMap.put(PigxQuartzEnum.SCHEDULE_JOB_KEY.getType(), sysJob);
|
||||
// 在jobGroup中包含租户ID以实现多租户隔离
|
||||
String jobGroupWithTenant = sysjob.getJobGroup() + "_tenant_" + tenantId;
|
||||
return TriggerKey.triggerKey(sysjob.getJobName(), jobGroupWithTenant);
|
||||
}
|
||||
|
||||
scheduler.triggerJob(getJobKey(sysJob), dataMap);
|
||||
}
|
||||
catch (SchedulerException e) {
|
||||
log.error("立刻执行定时任务,失败信息:{}", e.getMessage());
|
||||
return false;
|
||||
}
|
||||
/**
|
||||
* 添加或更新定时任务
|
||||
*
|
||||
* @param sysjob
|
||||
* @param scheduler
|
||||
*/
|
||||
public void addOrUpateJob(SysJob sysjob, Scheduler scheduler) {
|
||||
CronTrigger trigger = null;
|
||||
try {
|
||||
JobKey jobKey = getJobKey(sysjob);
|
||||
// 获得触发器
|
||||
TriggerKey triggerKey = getTriggerKey(sysjob);
|
||||
trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
|
||||
// 判断触发器是否存在(如果存在说明之前运行过但是在当前被禁用了,如果不存在说明一次都没运行过)
|
||||
if (trigger == null) {
|
||||
// 新建一个工作任务 指定任务类型为串接进行的
|
||||
JobDetail jobDetail = JobBuilder.newJob(PigxQuartzFactory.class).withIdentity(jobKey).build();
|
||||
// 将任务信息添加到任务信息中
|
||||
jobDetail.getJobDataMap().put(PigxQuartzEnum.SCHEDULE_JOB_KEY.getType(), sysjob);
|
||||
// 将cron表达式进行转换
|
||||
CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(sysjob.getCronExpression());
|
||||
cronScheduleBuilder = this.handleCronScheduleMisfirePolicy(sysjob, cronScheduleBuilder);
|
||||
// 创建触发器并将cron表达式对象给塞入
|
||||
trigger = TriggerBuilder.newTrigger()
|
||||
.withIdentity(triggerKey)
|
||||
.withSchedule(cronScheduleBuilder)
|
||||
.build();
|
||||
// 在调度器中将触发器和任务进行组合
|
||||
scheduler.scheduleJob(jobDetail, trigger);
|
||||
} else {
|
||||
CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(sysjob.getCronExpression());
|
||||
cronScheduleBuilder = this.handleCronScheduleMisfirePolicy(sysjob, cronScheduleBuilder);
|
||||
// 按照新的规则进行
|
||||
trigger = trigger.getTriggerBuilder()
|
||||
.withIdentity(triggerKey)
|
||||
.withSchedule(cronScheduleBuilder)
|
||||
.build();
|
||||
// 将任务信息更新到任务信息中
|
||||
trigger.getJobDataMap().put(PigxQuartzEnum.SCHEDULE_JOB_KEY.getType(), sysjob);
|
||||
// 重启
|
||||
scheduler.rescheduleJob(triggerKey, trigger);
|
||||
}
|
||||
// 如任务状态为暂停
|
||||
if (sysjob.getJobStatus().equals(PigxQuartzEnum.JOB_STATUS_NOT_RUNNING.getType())) {
|
||||
this.pauseJob(sysjob, scheduler);
|
||||
}
|
||||
} catch (SchedulerException e) {
|
||||
log.error("添加或更新定时任务,失败信息:{}", e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
/**
|
||||
* 立即执行一次任务
|
||||
*/
|
||||
public static boolean runOnce(Scheduler scheduler, SysJob sysJob) {
|
||||
try {
|
||||
// 参数
|
||||
JobDataMap dataMap = new JobDataMap();
|
||||
dataMap.put(PigxQuartzEnum.SCHEDULE_JOB_KEY.getType(), sysJob);
|
||||
|
||||
/**
|
||||
* 暂停定时任务
|
||||
* @param sysjob
|
||||
* @param scheduler
|
||||
*/
|
||||
public void pauseJob(SysJob sysjob, Scheduler scheduler) {
|
||||
try {
|
||||
if (scheduler != null) {
|
||||
scheduler.pauseJob(getJobKey(sysjob));
|
||||
}
|
||||
}
|
||||
catch (SchedulerException e) {
|
||||
log.error("暂停任务失败,失败信息:{}", e.getMessage());
|
||||
}
|
||||
scheduler.triggerJob(getJobKey(sysJob), dataMap);
|
||||
} catch (SchedulerException e) {
|
||||
log.error("立刻执行定时任务,失败信息:{}", e.getMessage());
|
||||
return false;
|
||||
}
|
||||
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* 恢复定时任务
|
||||
* @param sysjob
|
||||
* @param scheduler
|
||||
*/
|
||||
public void resumeJob(SysJob sysjob, Scheduler scheduler) {
|
||||
try {
|
||||
if (scheduler != null) {
|
||||
scheduler.resumeJob(getJobKey(sysjob));
|
||||
}
|
||||
}
|
||||
catch (SchedulerException e) {
|
||||
log.error("恢复任务失败,失败信息:{}", e.getMessage());
|
||||
}
|
||||
/**
|
||||
* 暂停定时任务
|
||||
*
|
||||
* @param sysjob
|
||||
* @param scheduler
|
||||
*/
|
||||
public void pauseJob(SysJob sysjob, Scheduler scheduler) {
|
||||
try {
|
||||
if (scheduler != null) {
|
||||
scheduler.pauseJob(getJobKey(sysjob));
|
||||
}
|
||||
} catch (SchedulerException e) {
|
||||
log.error("暂停任务失败,失败信息:{}", e.getMessage());
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 移除定时任务
|
||||
* @param sysjob
|
||||
* @param scheduler
|
||||
*/
|
||||
public void removeJob(SysJob sysjob, Scheduler scheduler) {
|
||||
try {
|
||||
if (scheduler != null) {
|
||||
// 停止触发器
|
||||
scheduler.pauseTrigger(getTriggerKey(sysjob));
|
||||
// 移除触发器
|
||||
scheduler.unscheduleJob(getTriggerKey(sysjob));
|
||||
// 删除任务
|
||||
scheduler.deleteJob(getJobKey(sysjob));
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
log.error("移除定时任务失败,失败信息:{}", e.getMessage());
|
||||
}
|
||||
}
|
||||
/**
|
||||
* 恢复定时任务
|
||||
*
|
||||
* @param sysjob
|
||||
* @param scheduler
|
||||
*/
|
||||
public void resumeJob(SysJob sysjob, Scheduler scheduler) {
|
||||
try {
|
||||
if (scheduler != null) {
|
||||
scheduler.resumeJob(getJobKey(sysjob));
|
||||
}
|
||||
} catch (SchedulerException e) {
|
||||
log.error("恢复任务失败,失败信息:{}", e.getMessage());
|
||||
}
|
||||
|
||||
/**
|
||||
* 启动所有运行定时任务
|
||||
* @param scheduler
|
||||
*/
|
||||
public void startJobs(Scheduler scheduler) {
|
||||
try {
|
||||
if (scheduler != null) {
|
||||
scheduler.resumeAll();
|
||||
}
|
||||
}
|
||||
catch (SchedulerException e) {
|
||||
log.error("启动所有运行定时任务失败,失败信息:{}", e.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 停止所有运行定时任务
|
||||
* @param scheduler
|
||||
*/
|
||||
public void pauseJobs(Scheduler scheduler) {
|
||||
try {
|
||||
if (scheduler != null) {
|
||||
scheduler.pauseAll();
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
log.error("暂停所有运行定时任务失败,失败信息:{}", e.getMessage());
|
||||
}
|
||||
}
|
||||
/**
|
||||
* 移除定时任务
|
||||
*
|
||||
* @param sysjob
|
||||
* @param scheduler
|
||||
*/
|
||||
public void removeJob(SysJob sysjob, Scheduler scheduler) {
|
||||
try {
|
||||
if (scheduler != null) {
|
||||
// 停止触发器
|
||||
scheduler.pauseTrigger(getTriggerKey(sysjob));
|
||||
// 移除触发器
|
||||
scheduler.unscheduleJob(getTriggerKey(sysjob));
|
||||
// 删除任务
|
||||
scheduler.deleteJob(getJobKey(sysjob));
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error("移除定时任务失败,失败信息:{}", e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取错失执行策略方法
|
||||
* @param sysJob
|
||||
* @param cronScheduleBuilder
|
||||
* @return
|
||||
*/
|
||||
private CronScheduleBuilder handleCronScheduleMisfirePolicy(SysJob sysJob,
|
||||
CronScheduleBuilder cronScheduleBuilder) {
|
||||
if (PigxQuartzEnum.MISFIRE_DEFAULT.getType().equals(sysJob.getMisfirePolicy())) {
|
||||
return cronScheduleBuilder;
|
||||
}
|
||||
else if (PigxQuartzEnum.MISFIRE_IGNORE_MISFIRES.getType().equals(sysJob.getMisfirePolicy())) {
|
||||
return cronScheduleBuilder.withMisfireHandlingInstructionIgnoreMisfires();
|
||||
}
|
||||
else if (PigxQuartzEnum.MISFIRE_FIRE_AND_PROCEED.getType().equals(sysJob.getMisfirePolicy())) {
|
||||
return cronScheduleBuilder.withMisfireHandlingInstructionFireAndProceed();
|
||||
}
|
||||
else if (PigxQuartzEnum.MISFIRE_DO_NOTHING.getType().equals(sysJob.getMisfirePolicy())) {
|
||||
return cronScheduleBuilder.withMisfireHandlingInstructionDoNothing();
|
||||
}
|
||||
else {
|
||||
return cronScheduleBuilder;
|
||||
}
|
||||
}
|
||||
/**
|
||||
* 启动所有运行定时任务
|
||||
*
|
||||
* @param scheduler
|
||||
*/
|
||||
public void startJobs(Scheduler scheduler) {
|
||||
try {
|
||||
if (scheduler != null) {
|
||||
scheduler.resumeAll();
|
||||
}
|
||||
} catch (SchedulerException e) {
|
||||
log.error("启动所有运行定时任务失败,失败信息:{}", e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断cron表达式是否正确
|
||||
* @param cronExpression
|
||||
* @return
|
||||
*/
|
||||
public boolean isValidCron(String cronExpression) {
|
||||
return CronExpression.isValidExpression(cronExpression);
|
||||
}
|
||||
/**
|
||||
* 停止所有运行定时任务
|
||||
*
|
||||
* @param scheduler
|
||||
*/
|
||||
public void pauseJobs(Scheduler scheduler) {
|
||||
try {
|
||||
if (scheduler != null) {
|
||||
scheduler.pauseAll();
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error("暂停所有运行定时任务失败,失败信息:{}", e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取错失执行策略方法
|
||||
*
|
||||
* @param sysJob
|
||||
* @param cronScheduleBuilder
|
||||
* @return
|
||||
*/
|
||||
private CronScheduleBuilder handleCronScheduleMisfirePolicy(SysJob sysJob,
|
||||
CronScheduleBuilder cronScheduleBuilder) {
|
||||
if (PigxQuartzEnum.MISFIRE_DEFAULT.getType().equals(sysJob.getMisfirePolicy())) {
|
||||
return cronScheduleBuilder;
|
||||
} else if (PigxQuartzEnum.MISFIRE_IGNORE_MISFIRES.getType().equals(sysJob.getMisfirePolicy())) {
|
||||
return cronScheduleBuilder.withMisfireHandlingInstructionIgnoreMisfires();
|
||||
} else if (PigxQuartzEnum.MISFIRE_FIRE_AND_PROCEED.getType().equals(sysJob.getMisfirePolicy())) {
|
||||
return cronScheduleBuilder.withMisfireHandlingInstructionFireAndProceed();
|
||||
} else if (PigxQuartzEnum.MISFIRE_DO_NOTHING.getType().equals(sysJob.getMisfirePolicy())) {
|
||||
return cronScheduleBuilder.withMisfireHandlingInstructionDoNothing();
|
||||
} else {
|
||||
return cronScheduleBuilder;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断cron表达式是否正确
|
||||
*
|
||||
* @param cronExpression
|
||||
* @return
|
||||
*/
|
||||
public boolean isValidCron(String cronExpression) {
|
||||
return CronExpression.isValidExpression(cronExpression);
|
||||
}
|
||||
|
||||
}
|
||||
Reference in new issue
Block a user