定时任务支持并发控制

ruoyi-quartz/src/main/java/com/ruoyi/quartz/util/AbstractQuartzJob.java

@@ -0,0 +1,108 @@
+package com.ruoyi.quartz.util;
+
+import java.util.Date;
+import org.quartz.Job;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import com.ruoyi.common.constant.Constants;
+import com.ruoyi.common.constant.ScheduleConstants;
+import com.ruoyi.common.utils.ExceptionUtil;
+import com.ruoyi.common.utils.StringUtils;
+import com.ruoyi.common.utils.bean.BeanUtils;
+import com.ruoyi.common.utils.spring.SpringUtils;
+import com.ruoyi.quartz.domain.SysJob;
+import com.ruoyi.quartz.domain.SysJobLog;
+import com.ruoyi.quartz.service.ISysJobLogService;
+
+/**
+ * 抽象quartz调用
+ *
+ * @author ruoyi
+ */
+public abstract class AbstractQuartzJob implements Job
+{
+    private static final Logger log = LoggerFactory.getLogger(AbstractQuartzJob.class);
+
+    /**
+     * 线程本地变量
+     */
+    private static ThreadLocal<Date> threadLocal = new ThreadLocal<>();
+
+    @Override
+    public void execute(JobExecutionContext context) throws JobExecutionException
+    {
+        SysJob sysJob = new SysJob();
+        BeanUtils.copyBeanProp(sysJob, context.getMergedJobDataMap().get(ScheduleConstants.TASK_PROPERTIES));
+        try
+        {
+            before(context, sysJob);
+            if (sysJob != null)
+            {
+                doExecute(context, sysJob);
+            }
+            after(context, sysJob, null);
+        }
+        catch (Exception e)
+        {
+            log.error("任务执行异常  - ：", e);
+            after(context, sysJob, e);
+        }
+    }
+
+    /**
+     * 执行前
+     *
+     * @param context 工作执行上下文对象
+     * @param sysJob 系统计划任务
+     */
+    protected void before(JobExecutionContext context, SysJob sysJob)
+    {
+        threadLocal.set(new Date());
+    }
+
+    /**
+     * 执行后
+     *
+     * @param context 工作执行上下文对象
+     * @param sysScheduleJob 系统计划任务
+     */
+    protected void after(JobExecutionContext context, SysJob sysJob, Exception e)
+    {
+        Date startTime = threadLocal.get();
+        threadLocal.remove();
+
+        final SysJobLog sysJobLog = new SysJobLog();
+        sysJobLog.setJobName(sysJob.getJobName());
+        sysJobLog.setJobGroup(sysJob.getJobGroup());
+        sysJobLog.setMethodName(sysJob.getMethodName());
+        sysJobLog.setMethodParams(sysJob.getMethodParams());
+        sysJobLog.setStartTime(startTime);
+        sysJobLog.setEndTime(new Date());
+        long runMs = sysJobLog.getEndTime().getTime() - sysJobLog.getStartTime().getTime();
+        sysJobLog.setJobMessage(sysJobLog.getJobName() + " 总共耗时：" + runMs + "毫秒");
+        if (e != null)
+        {
+            sysJobLog.setStatus(Constants.FAIL);
+            String errorMsg = StringUtils.substring(ExceptionUtil.getExceptionMessage(e), 0, 2000);
+            sysJobLog.setExceptionInfo(errorMsg);
+        }
+        else
+        {
+            sysJobLog.setStatus(Constants.SUCCESS);
+        }
+
+        // 写入数据库当中
+        SpringUtils.getBean(ISysJobLogService.class).addJobLog(sysJobLog);
+    }
+
+    /**
+     * 执行方法，由子类重载
+     *
+     * @param context 工作执行上下文对象
+     * @param sysJob 系统计划任务
+     * @throws Exception 执行过程中的异常
+     */
+    protected abstract void doExecute(JobExecutionContext context, SysJob sysJob) throws Exception;
+}

ruoyi-quartz/src/main/java/com/ruoyi/quartz/util/JobInvokeUtil.java

@@ -0,0 +1,57 @@
+package com.ruoyi.quartz.util;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import com.ruoyi.common.utils.StringUtils;
+import com.ruoyi.common.utils.spring.SpringUtils;
+import com.ruoyi.quartz.domain.SysJob;
+
+/**
+ * 任务执行工具
+ *
+ * @author ruoyi
+ */
+public class JobInvokeUtil
+{
+    /**
+     * 执行方法
+     *
+     * @param sysJob 系统任务
+     */
+    public static void invokeMethod(SysJob sysJob) throws Exception
+    {
+        Object bean = SpringUtils.getBean(sysJob.getJobName());
+        String methodName = sysJob.getMethodName();
+        String methodParams = sysJob.getMethodParams();
+
+        invokeSpringBean(bean, methodName, methodParams);
+    }
+
+    /**
+     * 调用任务方法
+     *
+     * @param bean 目标对象
+     * @param methodName 方法名称
+     * @param methodParams 方法参数
+     * @throws InvocationTargetException
+     * @throws SecurityException
+     * @throws NoSuchMethodException
+     * @throws IllegalArgumentException
+     * @throws IllegalAccessException
+     */
+    private static void invokeSpringBean(Object bean, String methodName, String methodParams)
+            throws NoSuchMethodException, SecurityException, IllegalAccessException, IllegalArgumentException,
+            InvocationTargetException
+    {
+        if (StringUtils.isNotEmpty(methodParams))
+        {
+            Method method = bean.getClass().getDeclaredMethod(methodName, String.class);
+            method.invoke(bean, methodParams);
+        }
+        else
+        {
+            Method method = bean.getClass().getDeclaredMethod(methodName);
+            method.invoke(bean);
+        }
+    }
+}

ruoyi-quartz/src/main/java/com/ruoyi/quartz/util/QuartzDisallowConcurrentExecution.java

@@ -0,0 +1,21 @@
+package com.ruoyi.quartz.util;
+
+import org.quartz.DisallowConcurrentExecution;
+import org.quartz.JobExecutionContext;
+import com.ruoyi.quartz.domain.SysJob;
+
+/**
+ * 定时任务处理（禁止并发执行）
+ * 
+ * @author ruoyi
+ *
+ */
+@DisallowConcurrentExecution
+public class QuartzDisallowConcurrentExecution extends AbstractQuartzJob
+{
+    @Override
+    protected void doExecute(JobExecutionContext context, SysJob sysJob) throws Exception
+    {
+        JobInvokeUtil.invokeMethod(sysJob);
+    }
+}

ruoyi-quartz/src/main/java/com/ruoyi/quartz/util/QuartzJobExecution.java

@@ -0,0 +1,19 @@
+package com.ruoyi.quartz.util;
+
+import org.quartz.JobExecutionContext;
+import com.ruoyi.quartz.domain.SysJob;
+
+/**
+ * 定时任务处理（允许并发执行）
+ * 
+ * @author ruoyi
+ *
+ */
+public class QuartzJobExecution extends AbstractQuartzJob
+{
+    @Override
+    protected void doExecute(JobExecutionContext context, SysJob sysJob) throws Exception
+    {
+        JobInvokeUtil.invokeMethod(sysJob);
+    }
+}

ruoyi-quartz/src/main/java/com/ruoyi/quartz/util/ScheduleJob.java

@@ -1,80 +0,0 @@
-package com.ruoyi.quartz.util;
-
-import java.util.Date;
-import java.util.concurrent.Future;
-import org.quartz.DisallowConcurrentExecution;
-import org.quartz.JobExecutionContext;
-import org.quartz.JobExecutionException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
-import org.springframework.scheduling.quartz.QuartzJobBean;
-import com.ruoyi.common.constant.Constants;
-import com.ruoyi.common.constant.ScheduleConstants;
-import com.ruoyi.common.utils.StringUtils;
-import com.ruoyi.common.utils.bean.BeanUtils;
-import com.ruoyi.common.utils.spring.SpringUtils;
-import com.ruoyi.quartz.domain.SysJob;
-import com.ruoyi.quartz.domain.SysJobLog;
-import com.ruoyi.quartz.service.ISysJobLogService;
-
-/**
- * 定时任务处理
- * 
- * @author ruoyi
- *
- */
-@DisallowConcurrentExecution
-public class ScheduleJob extends QuartzJobBean
-{
-    private static final Logger log = LoggerFactory.getLogger(ScheduleJob.class);
-
-    private ThreadPoolTaskExecutor executor = SpringUtils.getBean("threadPoolTaskExecutor");
-
-    private final static ISysJobLogService jobLogService = SpringUtils.getBean(ISysJobLogService.class);
-
-    @Override
-    protected void executeInternal(JobExecutionContext context) throws JobExecutionException
-    {
-        SysJob job = new SysJob();
-        BeanUtils.copyBeanProp(job, context.getMergedJobDataMap().get(ScheduleConstants.TASK_PROPERTIES));
-
-        SysJobLog jobLog = new SysJobLog();
-        jobLog.setJobName(job.getJobName());
-        jobLog.setJobGroup(job.getJobGroup());
-        jobLog.setMethodName(job.getMethodName());
-        jobLog.setMethodParams(job.getMethodParams());
-        jobLog.setCreateTime(new Date());
-
-        long startTime = System.currentTimeMillis();
-
-        try
-        {
-            // 执行任务
-            log.info("任务开始执行 - 名称：{} 方法：{}", job.getJobName(), job.getMethodName());
-            ScheduleRunnable task = new ScheduleRunnable(job.getJobName(), job.getMethodName(), job.getMethodParams());
-            Future<?> future = executor.submit(task);
-            future.get();
-            long times = System.currentTimeMillis() - startTime;
-            // 任务状态 0：成功 1：失败
-            jobLog.setStatus(Constants.SUCCESS);
-            jobLog.setJobMessage(job.getJobName() + " 总共耗时：" + times + "毫秒");
-
-            log.info("任务执行结束 - 名称：{} 耗时：{} 毫秒", job.getJobName(), times);
-        }
-        catch (Exception e)
-        {
-            log.info("任务执行失败 - 名称：{} 方法：{}", job.getJobName(), job.getMethodName());
-            log.error("任务执行异常  - ：", e);
-            long times = System.currentTimeMillis() - startTime;
-            jobLog.setJobMessage(job.getJobName() + " 总共耗时：" + times + "毫秒");
-            // 任务状态 0：成功 1：失败
-            jobLog.setStatus(Constants.FAIL);
-            jobLog.setExceptionInfo(StringUtils.substring(e.getMessage(), 0, 2000));
-        }
-        finally
-        {
-            jobLogService.addJobLog(jobLog);
-        }
-    }
-}

ruoyi-quartz/src/main/java/com/ruoyi/quartz/util/ScheduleRunnable.java

@@ -1,57 +0,0 @@
-package com.ruoyi.quartz.util;
-
-import java.lang.reflect.Method;
-import org.springframework.util.ReflectionUtils;
-import com.ruoyi.common.exception.BusinessException;
-import com.ruoyi.common.utils.StringUtils;
-import com.ruoyi.common.utils.spring.SpringUtils;
-
-/**
- * 执行定时任务
- * 
- * @author ruoyi
- *
- */
-public class ScheduleRunnable implements Runnable
-{
-    private Object target;
-    private Method method;
-    private String params;
-
-    public ScheduleRunnable(String beanName, String methodName, String params)
-            throws NoSuchMethodException, SecurityException
-    {
-        this.target = SpringUtils.getBean(beanName);
-        this.params = params;
-
-        if (StringUtils.isNotEmpty(params))
-        {
-            this.method = target.getClass().getDeclaredMethod(methodName, String.class);
-        }
-        else
-        {
-            this.method = target.getClass().getDeclaredMethod(methodName);
-        }
-    }
-
-    @Override
-    public void run()
-    {
-        try
-        {
-            ReflectionUtils.makeAccessible(method);
-            if (StringUtils.isNotEmpty(params))
-            {
-                method.invoke(target, params);
-            }
-            else
-            {
-                method.invoke(target);
-            }
-        }
-        catch (Exception e)
-        {
-            throw new BusinessException("执行定时任务失败", e);
-        }
-    }
-}

ruoyi-quartz/src/main/java/com/ruoyi/quartz/util/ScheduleUtils.java

@@ -2,6 +2,7 @@ package com.ruoyi.quartz.util;
 
 import org.quartz.CronScheduleBuilder;
 import org.quartz.CronTrigger;
+import org.quartz.Job;
 import org.quartz.JobBuilder;
 import org.quartz.JobDataMap;
 import org.quartz.JobDetail;
@@ -27,6 +28,18 @@ public class ScheduleUtils
 {
     private static final Logger log = LoggerFactory.getLogger(ScheduleUtils.class);
 
+    /**
+     * 得到quartz任务类
+     *
+     * @param sysJob 执行计划
+     * @return 具体执行任务类
+     */
+    private static Class<? extends Job> getQuartzJobClass(SysJob sysJob)
+    {
+        boolean isConcurrent = "0".equals(sysJob.getConcurrent());
+        return isConcurrent ? QuartzJobExecution.class : QuartzDisallowConcurrentExecution.class;
+    }
+
     /**
      * 获取触发器key
      */
@@ -62,147 +75,89 @@ public class ScheduleUtils
     /**
      * 创建定时任务
      */
-    public static void createScheduleJob(Scheduler scheduler, SysJob job)
+    public static void createScheduleJob(Scheduler scheduler, SysJob job) throws SchedulerException, TaskException
     {
-        try
+        Class<? extends Job> jobClass = getQuartzJobClass(job);
+        // 构建job信息
+        JobDetail jobDetail = JobBuilder.newJob(jobClass).withIdentity(getJobKey(job.getJobId())).build();
+
+        // 表达式调度构建器
+        CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(job.getCronExpression());
+        cronScheduleBuilder = handleCronScheduleMisfirePolicy(job, cronScheduleBuilder);
+
+        // 按新的cronExpression表达式构建一个新的trigger
+        CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity(getTriggerKey(job.getJobId()))
+                .withSchedule(cronScheduleBuilder).build();
+
+        // 放入参数，运行时的方法可以获取
+        jobDetail.getJobDataMap().put(ScheduleConstants.TASK_PROPERTIES, job);
+
+        scheduler.scheduleJob(jobDetail, trigger);
+
+        // 暂停任务
+        if (job.getStatus().equals(ScheduleConstants.Status.PAUSE.getValue()))
         {
-            // 构建job信息
-            JobDetail jobDetail = JobBuilder.newJob(ScheduleJob.class).withIdentity(getJobKey(job.getJobId())).build();
-
-            // 表达式调度构建器
-            CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(job.getCronExpression());
-            cronScheduleBuilder = handleCronScheduleMisfirePolicy(job, cronScheduleBuilder);
-
-            // 按新的cronExpression表达式构建一个新的trigger
-            CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity(getTriggerKey(job.getJobId()))
-                    .withSchedule(cronScheduleBuilder).build();
-
-            // 放入参数，运行时的方法可以获取
-            jobDetail.getJobDataMap().put(ScheduleConstants.TASK_PROPERTIES, job);
-
-            scheduler.scheduleJob(jobDetail, trigger);
-
-            // 暂停任务
-            if (job.getStatus().equals(ScheduleConstants.Status.PAUSE.getValue()))
-            {
-                pauseJob(scheduler, job.getJobId());
-            }
-        }
-        catch (SchedulerException e)
-        {
-            log.error("createScheduleJob 异常：", e);
-        }
-        catch (TaskException e)
-        {
-            log.error("createScheduleJob 异常：", e);
+            pauseJob(scheduler, job.getJobId());
         }
     }
 
     /**
      * 更新定时任务
      */
-    public static void updateScheduleJob(Scheduler scheduler, SysJob job)
+    public static void updateScheduleJob(Scheduler scheduler, SysJob job) throws SchedulerException, TaskException
     {
-        try
+        JobKey jobKey = getJobKey(job.getJobId());
+
+        // 判断是否存在
+        if (scheduler.checkExists(jobKey))
         {
-            TriggerKey triggerKey = getTriggerKey(job.getJobId());
-
-            // 表达式调度构建器
-            CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(job.getCronExpression());
-            cronScheduleBuilder = handleCronScheduleMisfirePolicy(job, cronScheduleBuilder);
-
-            CronTrigger trigger = getCronTrigger(scheduler, job.getJobId());
-
-            // 按新的cronExpression表达式重新构建trigger
-            trigger = trigger.getTriggerBuilder().withIdentity(triggerKey).withSchedule(cronScheduleBuilder).build();
-
-            // 参数
-            trigger.getJobDataMap().put(ScheduleConstants.TASK_PROPERTIES, job);
-
-            scheduler.rescheduleJob(triggerKey, trigger);
-
-            // 暂停任务
-            if (job.getStatus().equals(ScheduleConstants.Status.PAUSE.getValue()))
-            {
-                pauseJob(scheduler, job.getJobId());
-            }
-
+            // 先移除，然后做更新操作
+            scheduler.deleteJob(jobKey);
         }
-        catch (SchedulerException e)
+
+        createScheduleJob(scheduler, job);
+
+        // 暂停任务
+        if (job.getStatus().equals(ScheduleConstants.Status.PAUSE.getValue()))
         {
-            log.error("SchedulerException 异常：", e);
-        }
-        catch (TaskException e)
-        {
-            log.error("SchedulerException 异常：", e);
+            pauseJob(scheduler, job.getJobId());
         }
     }
 
     /**
      * 立即执行任务
      */
-    public static int run(Scheduler scheduler, SysJob job)
+    public static void run(Scheduler scheduler, SysJob job) throws SchedulerException
     {
-        int rows = 0;
-        try
-        {
-            // 参数
-            JobDataMap dataMap = new JobDataMap();
-            dataMap.put(ScheduleConstants.TASK_PROPERTIES, job);
+        // 参数
+        JobDataMap dataMap = new JobDataMap();
+        dataMap.put(ScheduleConstants.TASK_PROPERTIES, job);
 
-            scheduler.triggerJob(getJobKey(job.getJobId()), dataMap);
-            rows = 1;
-        }
-        catch (SchedulerException e)
-        {
-            log.error("run 异常：", e);
-        }
-        return rows;
+        scheduler.triggerJob(getJobKey(job.getJobId()), dataMap);
     }
 
     /**
      * 暂停任务
      */
-    public static void pauseJob(Scheduler scheduler, Long jobId)
+    public static void pauseJob(Scheduler scheduler, Long jobId) throws SchedulerException
     {
-        try
-        {
-            scheduler.pauseJob(getJobKey(jobId));
-        }
-        catch (SchedulerException e)
-        {
-            log.error("pauseJob 异常：", e);
-        }
+        scheduler.pauseJob(getJobKey(jobId));
     }
 
     /**
      * 恢复任务
      */
-    public static void resumeJob(Scheduler scheduler, Long jobId)
+    public static void resumeJob(Scheduler scheduler, Long jobId) throws SchedulerException
     {
-        try
-        {
-            scheduler.resumeJob(getJobKey(jobId));
-        }
-        catch (SchedulerException e)
-        {
-            log.error("resumeJob 异常：", e);
-        }
+        scheduler.resumeJob(getJobKey(jobId));
     }
 
     /**
      * 删除定时任务
      */
-    public static void deleteScheduleJob(Scheduler scheduler, Long jobId)
+    public static void deleteScheduleJob(Scheduler scheduler, Long jobId) throws SchedulerException
     {
-        try
-        {
-            scheduler.deleteJob(getJobKey(jobId));
-        }
-        catch (SchedulerException e)
-        {
-            log.error("deleteScheduleJob 异常：", e);
-        }
+        scheduler.deleteJob(getJobKey(jobId));
     }
 
     public static CronScheduleBuilder handleCronScheduleMisfirePolicy(SysJob job, CronScheduleBuilder cb)
@@ -219,7 +174,8 @@ public class ScheduleUtils
             case ScheduleConstants.MISFIRE_DO_NOTHING:
                 return cb.withMisfireHandlingInstructionDoNothing();
             default:
-                throw new TaskException("The task misfire policy '" + job.getMisfirePolicy() + "' cannot be used in cron schedule tasks", Code.CONFIG_ERROR);
+                throw new TaskException("The task misfire policy '" + job.getMisfirePolicy()
+                        + "' cannot be used in cron schedule tasks", Code.CONFIG_ERROR);
         }
     }
 }