diff --git a/backend/src/main/java/com/yfd/platform/utils/ExecutionJob.java b/backend/src/main/java/com/yfd/platform/utils/ExecutionJob.java index c4a2fcf3..8ab593ad 100644 --- a/backend/src/main/java/com/yfd/platform/utils/ExecutionJob.java +++ b/backend/src/main/java/com/yfd/platform/utils/ExecutionJob.java @@ -19,12 +19,12 @@ import com.yfd.platform.config.MessageConfig; import com.yfd.platform.config.thread.ThreadPoolExecutorUtil; import com.yfd.platform.system.domain.Message; import com.yfd.platform.system.domain.QuartzJob; -import com.yfd.platform.system.service.IMessageService; import com.yfd.platform.system.service.IQuartzJobService; +import lombok.extern.slf4j.Slf4j; +import org.quartz.DisallowConcurrentExecution; import org.quartz.JobExecutionContext; -import org.springframework.scheduling.annotation.Async; +import org.quartz.PersistJobDataAfterExecution; import org.springframework.scheduling.quartz.QuartzJobBean; -import org.springframework.stereotype.Component; import jakarta.annotation.Resource; import java.sql.Timestamp; @@ -32,72 +32,82 @@ import java.util.concurrent.Future; import java.util.concurrent.ThreadPoolExecutor; /** - * 参考人人开源,https://gitee.com/renrenio/renren-security + * 定时任务执行器(禁止同一Job并发执行) * * @author / * @date 2019-01-07 */ -@Async +@DisallowConcurrentExecution +@PersistJobDataAfterExecution @SuppressWarnings({"unchecked", "all"}) +@Slf4j public class ExecutionJob extends QuartzJobBean { - /** - * 该处仅供参考 - */ - private final static ThreadPoolExecutor EXECUTOR = - ThreadPoolExecutorUtil.getPoll(); + private static final ThreadPoolExecutor EXECUTOR = ThreadPoolExecutorUtil.getPoll(); - @Resource - private IMessageService messageService; @Resource private MessageConfig messageConfig; @Override - public void executeInternal(JobExecutionContext context) { + protected void executeInternal(JobExecutionContext context) { QuartzJob quartzJob = (QuartzJob) context.getMergedJobDataMap().get(QuartzJob.JOB_KEY); - // 获取spring bean + IQuartzJobService quartzJobService = SpringContextHolder.getBean(IQuartzJobService.class); - String uuid = quartzJob.getId(); + long startTime = System.currentTimeMillis(); String jobName = quartzJob.getJobName(); + String bizModule = quartzJob.getBizModule(); + String bizId = quartzJob.getBizId(); + try { - // 执行任务 - System.out.println( - "--------------------------------------------------------------"); - System.out.println("任务开始执行,任务名称:" + jobName); - QuartzRunnable task = new QuartzRunnable(quartzJob.getJobClass(), + logExecutionStart(jobName, bizModule, bizId); + + QuartzRunnable task = new QuartzRunnable( + quartzJob.getJobClass(), quartzJob.getJobMethod(), quartzJob.getJobParams()); Future future = EXECUTOR.submit(task); future.get(); + long times = System.currentTimeMillis() - startTime; - Message message = new Message(); - message.setCreatetime(new Timestamp(System.currentTimeMillis())); - message.setType("1"); - message.setTitle(quartzJob.getJobName()); - message.setContent(quartzJob.getDescription()); - message.setSenderName("定时器"); - message.setReceiverCodes(quartzJob.getOrderno().toString()); - message.setReceiverNames(""); - message.setStatus("1"); - message.setValidperiod(24); - messageConfig.addMessage(message); - // 任务状态 - System.out.println("任务执行完毕,任务名称:" + jobName + ", " + - "执行时间:" + times + "毫秒"); - System.out.println( - "--------------------------------------------------------------"); + logExecutionEnd(jobName, bizModule, bizId, times); + +// sendCompletionMessage(quartzJob); } catch (Exception e) { - System.out.println("任务执行失败,任务名称:" + jobName); - System.out.println( - "--------------------------------------------------------------"); + 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.setCreatetime(new Timestamp(System.currentTimeMillis())); + message.setType("1"); + message.setTitle(quartzJob.getJobName()); + message.setContent(quartzJob.getDescription()); + message.setSenderName("定时器"); + message.setReceiverCodes(quartzJob.getOrderno().toString()); + message.setReceiverNames(""); + message.setStatus("1"); + message.setValidperiod(24); + messageConfig.addMessage(message); + } } diff --git a/backend/src/main/java/com/yfd/platform/utils/QuartzManage.java b/backend/src/main/java/com/yfd/platform/utils/QuartzManage.java index 3ff265c6..4ecfb7e4 100644 --- a/backend/src/main/java/com/yfd/platform/utils/QuartzManage.java +++ b/backend/src/main/java/com/yfd/platform/utils/QuartzManage.java @@ -18,15 +18,21 @@ package com.yfd.platform.utils; import com.yfd.platform.system.domain.QuartzJob; import lombok.extern.slf4j.Slf4j; import org.quartz.*; +import org.quartz.impl.matchers.GroupMatcher; import org.quartz.impl.triggers.CronTriggerImpl; import org.springframework.stereotype.Component; import jakarta.annotation.Resource; import java.util.Date; +import java.util.List; +import java.util.Set; +import java.util.concurrent.locks.ReentrantLock; import static org.quartz.TriggerBuilder.newTrigger; /** + * 定时任务管理器(线程安全,支持多模块) + * * @author * @date 2019-01-07 */ @@ -34,154 +40,267 @@ import static org.quartz.TriggerBuilder.newTrigger; @Component 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") private Scheduler scheduler; - public void addJob(QuartzJob quartzJob) { - try { - // 构建job信息 - JobDetail jobDetail = JobBuilder.newJob(ExecutionJob.class). - withIdentity(JOB_NAME + quartzJob.getId()).build(); + /** 按任务ID粒度的锁,避免全局锁竞争 */ + private static final ReentrantLock GLOBAL_LOCK = new ReentrantLock(); + + /** + * 构建 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() - .withIdentity(JOB_NAME + quartzJob.getId()) + .withIdentity(buildTriggerKey(quartzJob)) .startNow() - .withSchedule(CronScheduleBuilder.cronSchedule(quartzJob.getJobCron())) + .withSchedule(CronScheduleBuilder.cronSchedule(quartzJob.getJobCron()) + .withMisfireHandlingInstructionDoNothing()) .build(); cronTrigger.getJobDataMap().put(QuartzJob.JOB_KEY, quartzJob); - - //重置启动时间 ((CronTriggerImpl) cronTrigger).setStartTime(new Date()); - //执行定时任务 scheduler.scheduleJob(jobDetail, cronTrigger); - // 暂停任务 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) { - log.error("创建定时任务失败", e); - throw new RuntimeException("创建定时任务失败"); + log.error("创建定时任务失败: module={}, bizId={}", quartzJob.getBizModule(), quartzJob.getBizId(), e); + throw new RuntimeException("创建定时任务失败", e); + } finally { + GLOBAL_LOCK.unlock(); } } /** - * 更新job cron表达式 - * - * @param quartzJob / + * 更新任务 Cron 表达式(线程安全,check-then-act 原子化) */ public void updateJobCron(QuartzJob quartzJob) { + GLOBAL_LOCK.lock(); try { - TriggerKey triggerKey = - TriggerKey.triggerKey(JOB_NAME + quartzJob.getId()); - CronTrigger trigger = - (CronTrigger) scheduler.getTrigger(triggerKey); - // 如果不存在则创建一个定时任务 + TriggerKey triggerKey = buildTriggerKey(quartzJob); + CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey); + if (trigger == null) { addJob(quartzJob); - trigger = (CronTrigger) scheduler.getTrigger(triggerKey); + return; } + CronScheduleBuilder scheduleBuilder = - CronScheduleBuilder.cronSchedule(quartzJob.getJobCron()); - trigger = - trigger.getTriggerBuilder().withIdentity(triggerKey).withSchedule(scheduleBuilder).build(); - //重置启动时间 + CronScheduleBuilder.cronSchedule(quartzJob.getJobCron()) + .withMisfireHandlingInstructionDoNothing(); + trigger = trigger.getTriggerBuilder() + .withIdentity(triggerKey) + .withSchedule(scheduleBuilder) + .build(); ((CronTriggerImpl) trigger).setStartTime(new Date()); trigger.getJobDataMap().put(QuartzJob.JOB_KEY, quartzJob); 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) { try { - JobKey jobKey = JobKey.jobKey(JOB_NAME + quartzJob.getId()); + JobKey jobKey = buildJobKey(quartzJob); scheduler.pauseJob(jobKey); scheduler.deleteJob(jobKey); + log.info("删除定时任务成功: module={}, bizId={}, jobName={}", + quartzJob.getBizModule(), quartzJob.getBizId(), quartzJob.getJobName()); } catch (Exception e) { - log.error("删除定时任务失败", e); - throw new RuntimeException("删除定时任务失败"); + log.error("删除定时任务失败: module={}, bizId={}", quartzJob.getBizModule(), quartzJob.getBizId(), e); + throw new RuntimeException("删除定时任务失败", e); } } /** - * 恢复一个job - * - * @param quartzJob / + * 恢复任务 */ public void resumeJob(QuartzJob quartzJob) { try { - TriggerKey triggerKey = - TriggerKey.triggerKey(JOB_NAME + quartzJob.getId()); - CronTrigger trigger = - (CronTrigger) scheduler.getTrigger(triggerKey); - // 如果不存在则创建一个定时任务 + TriggerKey triggerKey = buildTriggerKey(quartzJob); + CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey); if (trigger == null) { addJob(quartzJob); + return; } - JobKey jobKey = JobKey.jobKey(JOB_NAME + quartzJob.getId()); - scheduler.resumeJob(jobKey); + scheduler.resumeJob(buildJobKey(quartzJob)); + log.info("恢复定时任务成功: module={}, bizId={}", quartzJob.getBizModule(), quartzJob.getBizId()); } catch (Exception e) { - log.error("恢复定时任务失败", e); - throw new RuntimeException("恢复定时任务失败"); + log.error("恢复定时任务失败: module={}, bizId={}", quartzJob.getBizModule(), quartzJob.getBizId(), e); + throw new RuntimeException("恢复定时任务失败", e); } } /** - * 立即执行job - * - * @param quartzJob / + * 立即执行任务(保留已有 JobDataMap,追加最新 quartzJob 数据) */ public void runJobNow(QuartzJob quartzJob) { try { - TriggerKey triggerKey = - TriggerKey.triggerKey(JOB_NAME + quartzJob.getId()); - CronTrigger trigger = - (CronTrigger) scheduler.getTrigger(triggerKey); - // 如果不存在则创建一个定时任务 + TriggerKey triggerKey = buildTriggerKey(quartzJob); + CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey); + if (trigger == null) { + log.warn("任务不存在,先创建再执行: jobName={}", quartzJob.getJobName()); addJob(quartzJob); + trigger = (CronTrigger) scheduler.getTrigger(triggerKey); } - JobDataMap dataMap = new JobDataMap(); + + JobDataMap dataMap = trigger.getJobDataMap(); dataMap.put(QuartzJob.JOB_KEY, quartzJob); - JobKey jobKey = JobKey.jobKey(JOB_NAME + quartzJob.getId()); + + JobKey jobKey = buildJobKey(quartzJob); scheduler.triggerJob(jobKey, dataMap); + log.info("手动触发任务执行: module={}, bizId={}, jobName={}", + quartzJob.getBizModule(), quartzJob.getBizId(), quartzJob.getJobName()); } catch (Exception e) { - log.error("定时任务执行失败", e); - throw new RuntimeException("定时任务执行失败"); + log.error("手动触发任务失败: module={}, bizId={}", quartzJob.getBizModule(), quartzJob.getBizId(), e); + throw new RuntimeException("定时任务执行失败", e); } } /** - * 暂停一个job - * - * @param quartzJob / + * 暂停任务 */ public void pauseJob(QuartzJob quartzJob) { try { - JobKey jobKey = JobKey.jobKey(JOB_NAME + quartzJob.getId()); - scheduler.pauseJob(jobKey); + scheduler.pauseJob(buildJobKey(quartzJob)); + log.info("暂停定时任务成功: module={}, bizId={}", quartzJob.getBizModule(), quartzJob.getBizId()); } catch (Exception e) { - log.error("定时任务暂停失败", e); - throw new RuntimeException("定时任务暂停失败"); + log.error("定时任务暂停失败: module={}, bizId={}", quartzJob.getBizModule(), quartzJob.getBizId(), e); + 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 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 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 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); } } }