fix: 优化定时任务逻辑

This commit is contained in:
tangwei 2026-07-09 13:44:58 +08:00
parent b3d97c7cda
commit fd37d1c78e
2 changed files with 243 additions and 114 deletions

View File

@ -19,12 +19,12 @@ import com.yfd.platform.config.MessageConfig;
import com.yfd.platform.config.thread.ThreadPoolExecutorUtil; import com.yfd.platform.config.thread.ThreadPoolExecutorUtil;
import com.yfd.platform.system.domain.Message; import com.yfd.platform.system.domain.Message;
import com.yfd.platform.system.domain.QuartzJob; import com.yfd.platform.system.domain.QuartzJob;
import com.yfd.platform.system.service.IMessageService;
import com.yfd.platform.system.service.IQuartzJobService; import com.yfd.platform.system.service.IQuartzJobService;
import lombok.extern.slf4j.Slf4j;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.JobExecutionContext; import org.quartz.JobExecutionContext;
import org.springframework.scheduling.annotation.Async; import org.quartz.PersistJobDataAfterExecution;
import org.springframework.scheduling.quartz.QuartzJobBean; import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.stereotype.Component;
import jakarta.annotation.Resource; import jakarta.annotation.Resource;
import java.sql.Timestamp; import java.sql.Timestamp;
@ -32,48 +32,72 @@ import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.ThreadPoolExecutor;
/** /**
* 参考人人开源https://gitee.com/renrenio/renren-security * 定时任务执行器禁止同一Job并发执行
* *
* @author / * @author /
* @date 2019-01-07 * @date 2019-01-07
*/ */
@Async @DisallowConcurrentExecution
@PersistJobDataAfterExecution
@SuppressWarnings({"unchecked", "all"}) @SuppressWarnings({"unchecked", "all"})
@Slf4j
public class ExecutionJob extends QuartzJobBean { public class ExecutionJob extends QuartzJobBean {
/** private static final ThreadPoolExecutor EXECUTOR = ThreadPoolExecutorUtil.getPoll();
* 该处仅供参考
*/
private final static ThreadPoolExecutor EXECUTOR =
ThreadPoolExecutorUtil.getPoll();
@Resource
private IMessageService messageService;
@Resource @Resource
private MessageConfig messageConfig; private MessageConfig messageConfig;
@Override @Override
public void executeInternal(JobExecutionContext context) { protected void executeInternal(JobExecutionContext context) {
QuartzJob quartzJob = QuartzJob quartzJob =
(QuartzJob) context.getMergedJobDataMap().get(QuartzJob.JOB_KEY); (QuartzJob) context.getMergedJobDataMap().get(QuartzJob.JOB_KEY);
// 获取spring bean
IQuartzJobService quartzJobService = IQuartzJobService quartzJobService =
SpringContextHolder.getBean(IQuartzJobService.class); SpringContextHolder.getBean(IQuartzJobService.class);
String uuid = quartzJob.getId();
long startTime = System.currentTimeMillis(); long startTime = System.currentTimeMillis();
String jobName = quartzJob.getJobName(); String jobName = quartzJob.getJobName();
String bizModule = quartzJob.getBizModule();
String bizId = quartzJob.getBizId();
try { try {
// 执行任务 logExecutionStart(jobName, bizModule, bizId);
System.out.println(
"--------------------------------------------------------------"); QuartzRunnable task = new QuartzRunnable(
System.out.println("任务开始执行,任务名称:" + jobName); quartzJob.getJobClass(),
QuartzRunnable task = new QuartzRunnable(quartzJob.getJobClass(),
quartzJob.getJobMethod(), quartzJob.getJobMethod(),
quartzJob.getJobParams()); quartzJob.getJobParams());
Future<?> future = EXECUTOR.submit(task); Future<?> future = EXECUTOR.submit(task);
future.get(); future.get();
long times = System.currentTimeMillis() - startTime; long times = System.currentTimeMillis() - startTime;
logExecutionEnd(jobName, bizModule, bizId, times);
// sendCompletionMessage(quartzJob);
} catch (Exception e) {
log.error("任务执行失败: jobName={}, module={}, bizId={}", jobName, bizModule, bizId, e);
quartzJob.setStatus("0");
quartzJobService.updateById(quartzJob);
}
}
private void logExecutionStart(String jobName, String bizModule, String bizId) {
log.info("--------------------------------------------------------------");
log.info("任务开始执行,任务名称:" + jobName +
(bizModule != null ? ",模块:" + bizModule : "") +
(bizId != null ? "业务ID" + bizId : ""));
}
private void logExecutionEnd(String jobName, String bizModule, String bizId, long times) {
log.info("任务执行完毕,任务名称:" + jobName +
(bizModule != null ? ",模块:" + bizModule : "") +
",执行时间:" + times + "毫秒");
log.info("--------------------------------------------------------------");
}
private void sendCompletionMessage(QuartzJob quartzJob) {
Message message = new Message(); Message message = new Message();
message.setCreatetime(new Timestamp(System.currentTimeMillis())); message.setCreatetime(new Timestamp(System.currentTimeMillis()));
message.setType("1"); message.setType("1");
@ -85,19 +109,5 @@ public class ExecutionJob extends QuartzJobBean {
message.setStatus("1"); message.setStatus("1");
message.setValidperiod(24); message.setValidperiod(24);
messageConfig.addMessage(message); messageConfig.addMessage(message);
// 任务状态
System.out.println("任务执行完毕,任务名称:" + jobName + ", " +
"执行时间:" + times + "毫秒");
System.out.println(
"--------------------------------------------------------------");
} catch (Exception e) {
System.out.println("任务执行失败,任务名称:" + jobName);
System.out.println(
"--------------------------------------------------------------");
quartzJob.setStatus("0");
//更新状态
quartzJobService.updateById(quartzJob);
} }
} }
}

View File

@ -18,15 +18,21 @@ package com.yfd.platform.utils;
import com.yfd.platform.system.domain.QuartzJob; import com.yfd.platform.system.domain.QuartzJob;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.quartz.*; import org.quartz.*;
import org.quartz.impl.matchers.GroupMatcher;
import org.quartz.impl.triggers.CronTriggerImpl; import org.quartz.impl.triggers.CronTriggerImpl;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import jakarta.annotation.Resource; import jakarta.annotation.Resource;
import java.util.Date; import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.concurrent.locks.ReentrantLock;
import static org.quartz.TriggerBuilder.newTrigger; import static org.quartz.TriggerBuilder.newTrigger;
/** /**
* 定时任务管理器线程安全支持多模块
*
* @author * @author
* @date 2019-01-07 * @date 2019-01-07
*/ */
@ -34,154 +40,267 @@ import static org.quartz.TriggerBuilder.newTrigger;
@Component @Component
public class QuartzManage { public class QuartzManage {
private static final String JOB_NAME = "TASK_"; private static final String JOB_NAME_PREFIX = "TASK_";
private static final String GROUP_SEPARATOR = "_";
@Resource(name = "scheduler") @Resource(name = "scheduler")
private Scheduler scheduler; private Scheduler scheduler;
public void addJob(QuartzJob quartzJob) { /** 按任务ID粒度的锁避免全局锁竞争 */
try { private static final ReentrantLock GLOBAL_LOCK = new ReentrantLock();
// 构建job信息
JobDetail jobDetail = JobBuilder.newJob(ExecutionJob.class). /**
withIdentity(JOB_NAME + quartzJob.getId()).build(); * 构建 JobKey包含 bizModule 用于分组隔离
*/
private JobKey buildJobKey(QuartzJob quartzJob) {
String group = resolveGroup(quartzJob);
return JobKey.jobKey(JOB_NAME_PREFIX + quartzJob.getId(), group);
}
/**
* 构建 TriggerKey
*/
private TriggerKey buildTriggerKey(QuartzJob quartzJob) {
String group = resolveGroup(quartzJob);
return TriggerKey.triggerKey(JOB_NAME_PREFIX + quartzJob.getId(), group);
}
/**
* 解析 Quartz 分组名 bizModule 则按模块分组否则默认分组
*/
private String resolveGroup(QuartzJob quartzJob) {
if (quartzJob.getBizModule() != null && !quartzJob.getBizModule().isBlank()) {
return quartzJob.getBizModule();
}
return "DEFAULT";
}
/**
* 添加定时任务线程安全
*/
public void addJob(QuartzJob quartzJob) {
GLOBAL_LOCK.lock();
try {
JobDetail jobDetail = JobBuilder.newJob(ExecutionJob.class)
.withIdentity(buildJobKey(quartzJob))
.storeDurably()
.build();
//通过触发器名和cron 表达式创建 Trigger
Trigger cronTrigger = newTrigger() Trigger cronTrigger = newTrigger()
.withIdentity(JOB_NAME + quartzJob.getId()) .withIdentity(buildTriggerKey(quartzJob))
.startNow() .startNow()
.withSchedule(CronScheduleBuilder.cronSchedule(quartzJob.getJobCron())) .withSchedule(CronScheduleBuilder.cronSchedule(quartzJob.getJobCron())
.withMisfireHandlingInstructionDoNothing())
.build(); .build();
cronTrigger.getJobDataMap().put(QuartzJob.JOB_KEY, quartzJob); cronTrigger.getJobDataMap().put(QuartzJob.JOB_KEY, quartzJob);
//重置启动时间
((CronTriggerImpl) cronTrigger).setStartTime(new Date()); ((CronTriggerImpl) cronTrigger).setStartTime(new Date());
//执行定时任务
scheduler.scheduleJob(jobDetail, cronTrigger); scheduler.scheduleJob(jobDetail, cronTrigger);
// 暂停任务
if ("0".equals(quartzJob.getStatus())) { if ("0".equals(quartzJob.getStatus())) {
pauseJob(quartzJob); scheduler.pauseJob(buildJobKey(quartzJob));
} }
log.info("创建定时任务成功: module={}, bizId={}, jobName={}",
quartzJob.getBizModule(), quartzJob.getBizId(), quartzJob.getJobName());
} catch (Exception e) { } catch (Exception e) {
log.error("创建定时任务失败", e); log.error("创建定时任务失败: module={}, bizId={}", quartzJob.getBizModule(), quartzJob.getBizId(), e);
throw new RuntimeException("创建定时任务失败"); throw new RuntimeException("创建定时任务失败", e);
} finally {
GLOBAL_LOCK.unlock();
} }
} }
/** /**
* 更新job cron表达式 * 更新任务 Cron 表达式线程安全check-then-act 原子化
*
* @param quartzJob /
*/ */
public void updateJobCron(QuartzJob quartzJob) { public void updateJobCron(QuartzJob quartzJob) {
GLOBAL_LOCK.lock();
try { try {
TriggerKey triggerKey = TriggerKey triggerKey = buildTriggerKey(quartzJob);
TriggerKey.triggerKey(JOB_NAME + quartzJob.getId()); CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
CronTrigger trigger =
(CronTrigger) scheduler.getTrigger(triggerKey);
// 如果不存在则创建一个定时任务
if (trigger == null) { if (trigger == null) {
addJob(quartzJob); addJob(quartzJob);
trigger = (CronTrigger) scheduler.getTrigger(triggerKey); return;
} }
CronScheduleBuilder scheduleBuilder = CronScheduleBuilder scheduleBuilder =
CronScheduleBuilder.cronSchedule(quartzJob.getJobCron()); CronScheduleBuilder.cronSchedule(quartzJob.getJobCron())
trigger = .withMisfireHandlingInstructionDoNothing();
trigger.getTriggerBuilder().withIdentity(triggerKey).withSchedule(scheduleBuilder).build(); trigger = trigger.getTriggerBuilder()
//重置启动时间 .withIdentity(triggerKey)
.withSchedule(scheduleBuilder)
.build();
((CronTriggerImpl) trigger).setStartTime(new Date()); ((CronTriggerImpl) trigger).setStartTime(new Date());
trigger.getJobDataMap().put(QuartzJob.JOB_KEY, quartzJob); trigger.getJobDataMap().put(QuartzJob.JOB_KEY, quartzJob);
scheduler.rescheduleJob(triggerKey, trigger); scheduler.rescheduleJob(triggerKey, trigger);
// 暂停任务
if ("0".equals(quartzJob.getStatus())) {
pauseJob(quartzJob);
}
} catch (Exception e) {
log.error("更新定时任务失败", e);
throw new RuntimeException("更新定时任务失败");
}
if ("0".equals(quartzJob.getStatus())) {
scheduler.pauseJob(buildJobKey(quartzJob));
} else {
scheduler.resumeJob(buildJobKey(quartzJob));
}
log.info("更新定时任务成功: module={}, bizId={}, jobName={}",
quartzJob.getBizModule(), quartzJob.getBizId(), quartzJob.getJobName());
} catch (Exception e) {
log.error("更新定时任务失败: module={}, bizId={}", quartzJob.getBizModule(), quartzJob.getBizId(), e);
throw new RuntimeException("更新定时任务失败", e);
} finally {
GLOBAL_LOCK.unlock();
}
} }
/** /**
* 删除一个job * 删除任务
*
* @param quartzJob /
*/ */
public void deleteJob(QuartzJob quartzJob) { public void deleteJob(QuartzJob quartzJob) {
try { try {
JobKey jobKey = JobKey.jobKey(JOB_NAME + quartzJob.getId()); JobKey jobKey = buildJobKey(quartzJob);
scheduler.pauseJob(jobKey); scheduler.pauseJob(jobKey);
scheduler.deleteJob(jobKey); scheduler.deleteJob(jobKey);
log.info("删除定时任务成功: module={}, bizId={}, jobName={}",
quartzJob.getBizModule(), quartzJob.getBizId(), quartzJob.getJobName());
} catch (Exception e) { } catch (Exception e) {
log.error("删除定时任务失败", e); log.error("删除定时任务失败: module={}, bizId={}", quartzJob.getBizModule(), quartzJob.getBizId(), e);
throw new RuntimeException("删除定时任务失败"); throw new RuntimeException("删除定时任务失败", e);
} }
} }
/** /**
* 恢复一个job * 恢复任务
*
* @param quartzJob /
*/ */
public void resumeJob(QuartzJob quartzJob) { public void resumeJob(QuartzJob quartzJob) {
try { try {
TriggerKey triggerKey = TriggerKey triggerKey = buildTriggerKey(quartzJob);
TriggerKey.triggerKey(JOB_NAME + quartzJob.getId()); CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
CronTrigger trigger =
(CronTrigger) scheduler.getTrigger(triggerKey);
// 如果不存在则创建一个定时任务
if (trigger == null) { if (trigger == null) {
addJob(quartzJob); addJob(quartzJob);
return;
} }
JobKey jobKey = JobKey.jobKey(JOB_NAME + quartzJob.getId()); scheduler.resumeJob(buildJobKey(quartzJob));
scheduler.resumeJob(jobKey); log.info("恢复定时任务成功: module={}, bizId={}", quartzJob.getBizModule(), quartzJob.getBizId());
} catch (Exception e) { } catch (Exception e) {
log.error("恢复定时任务失败", e); log.error("恢复定时任务失败: module={}, bizId={}", quartzJob.getBizModule(), quartzJob.getBizId(), e);
throw new RuntimeException("恢复定时任务失败"); throw new RuntimeException("恢复定时任务失败", e);
} }
} }
/** /**
* 立即执行job * 立即执行任务保留已有 JobDataMap追加最新 quartzJob 数据
*
* @param quartzJob /
*/ */
public void runJobNow(QuartzJob quartzJob) { public void runJobNow(QuartzJob quartzJob) {
try { try {
TriggerKey triggerKey = TriggerKey triggerKey = buildTriggerKey(quartzJob);
TriggerKey.triggerKey(JOB_NAME + quartzJob.getId()); CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
CronTrigger trigger =
(CronTrigger) scheduler.getTrigger(triggerKey);
// 如果不存在则创建一个定时任务
if (trigger == null) { if (trigger == null) {
log.warn("任务不存在,先创建再执行: jobName={}", quartzJob.getJobName());
addJob(quartzJob); addJob(quartzJob);
trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
} }
JobDataMap dataMap = new JobDataMap();
JobDataMap dataMap = trigger.getJobDataMap();
dataMap.put(QuartzJob.JOB_KEY, quartzJob); dataMap.put(QuartzJob.JOB_KEY, quartzJob);
JobKey jobKey = JobKey.jobKey(JOB_NAME + quartzJob.getId());
JobKey jobKey = buildJobKey(quartzJob);
scheduler.triggerJob(jobKey, dataMap); scheduler.triggerJob(jobKey, dataMap);
log.info("手动触发任务执行: module={}, bizId={}, jobName={}",
quartzJob.getBizModule(), quartzJob.getBizId(), quartzJob.getJobName());
} catch (Exception e) { } catch (Exception e) {
log.error("定时任务执行失败", e); log.error("手动触发任务失败: module={}, bizId={}", quartzJob.getBizModule(), quartzJob.getBizId(), e);
throw new RuntimeException("定时任务执行失败"); throw new RuntimeException("定时任务执行失败", e);
} }
} }
/** /**
* 暂停一个job * 暂停任务
*
* @param quartzJob /
*/ */
public void pauseJob(QuartzJob quartzJob) { public void pauseJob(QuartzJob quartzJob) {
try { try {
JobKey jobKey = JobKey.jobKey(JOB_NAME + quartzJob.getId()); scheduler.pauseJob(buildJobKey(quartzJob));
scheduler.pauseJob(jobKey); log.info("暂停定时任务成功: module={}, bizId={}", quartzJob.getBizModule(), quartzJob.getBizId());
} catch (Exception e) { } catch (Exception e) {
log.error("定时任务暂停失败", e); log.error("定时任务暂停失败: module={}, bizId={}", quartzJob.getBizModule(), quartzJob.getBizId(), e);
throw new RuntimeException("定时任务暂停失败"); throw new RuntimeException("定时任务暂停失败", e);
}
}
// ==================== 多模块支持方法 ====================
/**
* 根据业务模块和业务ID查找已存在的任务
*
* @param bizModule 业务模块编码
* @param bizId 业务主键ID
* @return 是否存在
*/
public boolean existsByBiz(String bizModule, String bizId) {
if (bizModule == null || bizModule.isBlank() || bizId == null || bizId.isBlank()) {
return false;
}
try {
Set<JobKey> jobKeys = scheduler.getJobKeys(GroupMatcher.groupEquals(bizModule));
for (JobKey jobKey : jobKeys) {
JobDetail detail = scheduler.getJobDetail(jobKey);
if (detail != null) {
QuartzJob job = (QuartzJob) detail.getJobDataMap().get(QuartzJob.JOB_KEY);
if (job != null && bizId.equals(job.getBizId())) {
return true;
}
}
}
} catch (SchedulerException e) {
log.error("查询任务失败: module={}, bizId={}", bizModule, bizId, e);
}
return false;
}
/**
* 按模块暂停该模块下所有任务
*/
public void pauseAllByModule(String bizModule) {
if (bizModule == null || bizModule.isBlank()) {
return;
}
try {
Set<JobKey> jobKeys = scheduler.getJobKeys(GroupMatcher.groupEquals(bizModule));
scheduler.pauseJobs(org.quartz.impl.matchers.GroupMatcher.groupEquals(bizModule));
log.info("暂停模块[{}]下所有任务,共{}个", bizModule, jobKeys.size());
} catch (SchedulerException e) {
log.error("暂停模块任务失败: module={}", bizModule, e);
}
}
/**
* 按模块恢复该模块下所有任务
*/
public void resumeAllByModule(String bizModule) {
if (bizModule == null || bizModule.isBlank()) {
return;
}
try {
scheduler.resumeJobs(org.quartz.impl.matchers.GroupMatcher.groupEquals(bizModule));
log.info("恢复模块[{}]下所有任务", bizModule);
} catch (SchedulerException e) {
log.error("恢复模块任务失败: module={}", bizModule, e);
}
}
/**
* 按模块删除该模块下所有任务
*/
public void deleteAllByModule(String bizModule) {
if (bizModule == null || bizModule.isBlank()) {
return;
}
try {
Set<JobKey> jobKeys = scheduler.getJobKeys(GroupMatcher.groupEquals(bizModule));
scheduler.deleteJobs(new java.util.ArrayList<>(jobKeys));
log.info("删除模块[{}]下所有任务,共{}个", bizModule, jobKeys.size());
} catch (SchedulerException e) {
log.error("删除模块任务失败: module={}", bizModule, e);
} }
} }
} }