定时任务并发串行设定提交

Signed-off-by: TanYibin <5491541@qq.com>

定时任务并发串行设定提交
Signed-off-by: TanYibin <5491541@qq.com>
谭毅彬
1 parent 4be4b653
Showing 8 changed files with 110 additions and 129 deletions
huaheng-wms-core/src/main/java/org/jeecg/modules/message/job/SendMsgJob.java
huaheng-wms-core/src/main/java/org/jeecg/modules/quartz/job/AsyncJob.java
huaheng-wms-core/src/main/java/org/jeecg/modules/quartz/job/SampleJob.java
huaheng-wms-core/src/main/java/org/jeecg/modules/quartz/job/SampleParamJob.java
huaheng-wms-core/src/main/java/org/jeecg/modules/wms/monitor/job/ErpTask.java
huaheng-wms-core/src/main/java/org/jeecg/modules/wms/monitor/job/MesTask.java
huaheng-wms-core/src/main/java/org/jeecg/modules/wms/monitor/job/MesTryTask.java
huaheng-wms-core/src/main/java/org/jeecg/modules/wms/monitor/job/WcsTask.java
@@ -8,9 +8,11 @@ import org.jeecg.modules.message.handle.ISendMsgHandle;
 import org.jeecg.modules.message.handle.enums.SendMsgStatusEnum;
 import org.jeecg.modules.message.handle.enums.SendMsgTypeEnum;
 import org.jeecg.modules.message.service.ISysMessageService;
+import org.quartz.DisallowConcurrentExecution;
 import org.quartz.Job;
 import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
+import org.quartz.PersistJobDataAfterExecution;
 import org.springframework.beans.factory.annotation.Autowired;
 import com.alibaba.fastjson.JSON;
@@ -23,6 +25,8 @@ import lombok.extern.slf4j.Slf4j;
  */
 @Slf4j
+@PersistJobDataAfterExecution
+@DisallowConcurrentExecution
 public class SendMsgJob implements Job {
     @Autowired
@@ -11,9 +11,9 @@ import org.quartz.*;
  * @author:      taoyan
  * @date:        2020年06月19日
  */
+@Slf4j
 @PersistJobDataAfterExecution
 @DisallowConcurrentExecution
-@Slf4j
 public class AsyncJob implements Job {
     @Override
 package org.jeecg.modules.quartz.job;
 import org.jeecg.common.util.DateUtils;
+import org.quartz.DisallowConcurrentExecution;
 import org.quartz.Job;
 import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
+import org.quartz.PersistJobDataAfterExecution;
 import lombok.extern.slf4j.Slf4j;
@@ -12,11 +14,13 @@ import lombok.extern.slf4j.Slf4j;
  * @Author Scott
  */
 @Slf4j
+@PersistJobDataAfterExecution
+@DisallowConcurrentExecution
 public class SampleJob implements Job {
     @Override
     public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
         log.info(" Job Execution key：" + jobExecutionContext.getJobDetail().getKey());
-        log.info(String.format(" Jeecg-Boot 普通定时任务 SampleJob !  时间:" + DateUtils.getTimestamp()));
+        log.info(String.format(" Jeecg-Boot 普通定时任务 SampleJob! 时间:" + DateUtils.getTimestamp()));
     }
 }
 package org.jeecg.modules.quartz.job;
 import org.jeecg.common.util.DateUtils;
+import org.quartz.DisallowConcurrentExecution;
 import org.quartz.Job;
 import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
+import org.quartz.PersistJobDataAfterExecution;
 import lombok.extern.slf4j.Slf4j;
@@ -12,6 +14,8 @@ import lombok.extern.slf4j.Slf4j;
  * @Author Scott
  */
 @Slf4j
+@PersistJobDataAfterExecution
+@DisallowConcurrentExecution
 public class SampleParamJob implements Job {
     /**
@@ -26,6 +30,6 @@ public class SampleParamJob implements Job {
     @Override
     public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
         log.info(" Job Execution key：" + jobExecutionContext.getJobDetail().getKey());
-        log.info(String.format("welcome %s! Jeecg-Boot 带参数定时任务 SampleParamJob !   时间:" + DateUtils.now(), this.parameter));
+        log.info(String.format("welcome %s! Jeecg-Boot 带参数定时任务 SampleParamJob! 时间:" + DateUtils.now(), this.parameter));
     }
 }
@@ -13,9 +13,11 @@ import org.jeecg.modules.wms.receipt.receiptHeader.service.IReceiptHeaderService
 import org.jeecg.modules.wms.shipment.shipmentHeader.entity.ShipmentHeader;
 import org.jeecg.modules.wms.shipment.shipmentHeader.service.IShipmentHeaderService;
 import org.jeecg.utils.constant.QuantityConstant;
+import org.quartz.DisallowConcurrentExecution;
 import org.quartz.Job;
 import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
+import org.quartz.PersistJobDataAfterExecution;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
@@ -28,52 +30,43 @@ import lombok.extern.slf4j.Slf4j;
  */
 @Slf4j
+@PersistJobDataAfterExecution
+@DisallowConcurrentExecution
 public class ErpTask implements Job {
     @Resource
     private IReceiptHeaderService receiptHeaderService;
+
     @Resource
     private IShipmentHeaderService shipmentHeaderService;
+
     @Resource
     private IErpService erpService;
-    // 并发控制
-    Map<String, Boolean> runningTaskMap = new HashMap<>();
     @Override
     public void execute(JobExecutionContext context) throws JobExecutionException {
-        String taskKey = "ErpTask";
-        if (MapUtils.getBoolean(runningTaskMap, taskKey, false)) {
-            return;
-        }
-        try {
-            runningTaskMap.put(taskKey, true);
-            LambdaQueryWrapper<ReceiptHeader> receiptHeaderLambdaQueryWrapper = Wrappers.lambdaQuery();
-            receiptHeaderLambdaQueryWrapper.eq(ReceiptHeader::getFirstStatus, QuantityConstant.RECEIPT_HEADER_COMPLETED).eq(ReceiptHeader::getLastStatus,
-                QuantityConstant.RECEIPT_HEADER_COMPLETED);
-            List<ReceiptHeader> receiptHeaderList = receiptHeaderService.list(receiptHeaderLambdaQueryWrapper);
-            for (ReceiptHeader receiptHeader : receiptHeaderList) {
-                try {
-                    erpService.backReceipt(receiptHeader);
-                } catch (Exception e) {
-                    e.printStackTrace();
-                }
+        LambdaQueryWrapper<ReceiptHeader> receiptHeaderLambdaQueryWrapper = Wrappers.lambdaQuery();
+        receiptHeaderLambdaQueryWrapper.eq(ReceiptHeader::getFirstStatus, QuantityConstant.RECEIPT_HEADER_COMPLETED).eq(ReceiptHeader::getLastStatus,
+            QuantityConstant.RECEIPT_HEADER_COMPLETED);
+        List<ReceiptHeader> receiptHeaderList = receiptHeaderService.list(receiptHeaderLambdaQueryWrapper);
+        for (ReceiptHeader receiptHeader : receiptHeaderList) {
+            try {
+                erpService.backReceipt(receiptHeader);
+            } catch (Exception e) {
+                e.printStackTrace();
             }
+        }
-            LambdaQueryWrapper<ShipmentHeader> shipmentHeaderLambdaQueryWrapper = Wrappers.lambdaQuery();
-            shipmentHeaderLambdaQueryWrapper.eq(ShipmentHeader::getFirstStatus, QuantityConstant.RECEIPT_HEADER_COMPLETED).eq(ShipmentHeader::getLastStatus,
-                QuantityConstant.RECEIPT_HEADER_COMPLETED);
-            List<ShipmentHeader> shipmentHeaderList = shipmentHeaderService.list(shipmentHeaderLambdaQueryWrapper);
-            for (ShipmentHeader shipmentHeader : shipmentHeaderList) {
-                try {
-                    erpService.backShipment(shipmentHeader);
-                } catch (Exception e) {
-                    e.printStackTrace();
-                }
+        LambdaQueryWrapper<ShipmentHeader> shipmentHeaderLambdaQueryWrapper = Wrappers.lambdaQuery();
+        shipmentHeaderLambdaQueryWrapper.eq(ShipmentHeader::getFirstStatus, QuantityConstant.RECEIPT_HEADER_COMPLETED).eq(ShipmentHeader::getLastStatus,
+            QuantityConstant.RECEIPT_HEADER_COMPLETED);
+        List<ShipmentHeader> shipmentHeaderList = shipmentHeaderService.list(shipmentHeaderLambdaQueryWrapper);
+        for (ShipmentHeader shipmentHeader : shipmentHeaderList) {
+            try {
+                erpService.backShipment(shipmentHeader);
+            } catch (Exception e) {
+                e.printStackTrace();
             }
-        } catch (Exception e) {
-            e.printStackTrace();
-        } finally {
-            runningTaskMap.put(taskKey, false);
         }
     }
 }
@@ -11,9 +11,11 @@ import org.jeecg.modules.wms.api.mes.servuce.IMesService;
 import org.jeecg.modules.wms.task.taskHeader.entity.TaskHeader;
 import org.jeecg.modules.wms.task.taskHeader.service.ITaskHeaderService;
 import org.jeecg.utils.constant.QuantityConstant;
+import org.quartz.DisallowConcurrentExecution;
 import org.quartz.Job;
 import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
+import org.quartz.PersistJobDataAfterExecution;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
@@ -24,14 +26,15 @@ import lombok.extern.slf4j.Slf4j;
  * @author 游杰
  */
 @Slf4j
+@PersistJobDataAfterExecution
+@DisallowConcurrentExecution
 public class MesTask implements Job {
     @Resource
     private IMesService mesService;
+
     @Resource
     private ITaskHeaderService taskHeaderService;
-    // 并发控制
-    Map<String, Boolean> runningTaskMap = new HashMap<>();
     /**
      * 找到完成状态的，没有回传成功的所有任务，然后回传信息给MES
@@ -40,43 +43,32 @@ public class MesTask implements Job {
      */
     @Override
     public void execute(JobExecutionContext context) throws JobExecutionException {
-        String taskKey = "MesTask";
-        if (MapUtils.getBoolean(runningTaskMap, taskKey, false)) {
-            return;
-        }
-        try {
-            runningTaskMap.put(taskKey, true);
-            LambdaQueryWrapper<TaskHeader> taskHeaderLambdaQueryWrapper = Wrappers.lambdaQuery();
-            taskHeaderLambdaQueryWrapper.eq(TaskHeader::getStatus, QuantityConstant.TASK_STATUS_COMPLETED).ne(TaskHeader::getBackMes, QuantityConstant.MES_SEND_OK)
-                .in(TaskHeader::getTaskType, QuantityConstant.TASK_TYPE_WHOLERECEIPT, QuantityConstant.TASK_TYPE_SUPPLEMENTRECEIPT)
-                .le(TaskHeader::getBackMesTimes, QuantityConstant.MAX_MES_TRY_TIMES);
-            List<TaskHeader> taskHeaderList = taskHeaderService.list(taskHeaderLambdaQueryWrapper);
-            for (TaskHeader taskHeader : taskHeaderList) {
-                try {
-                    taskHeaderService.addMesTryTimes(taskHeader);
-                    mesService.backMesReceipt(taskHeader);
-                } catch (Exception e) {
-                    e.printStackTrace();
-                }
+        LambdaQueryWrapper<TaskHeader> taskHeaderLambdaQueryWrapper = Wrappers.lambdaQuery();
+        taskHeaderLambdaQueryWrapper.eq(TaskHeader::getStatus, QuantityConstant.TASK_STATUS_COMPLETED).ne(TaskHeader::getBackMes, QuantityConstant.MES_SEND_OK)
+            .in(TaskHeader::getTaskType, QuantityConstant.TASK_TYPE_WHOLERECEIPT, QuantityConstant.TASK_TYPE_SUPPLEMENTRECEIPT)
+            .le(TaskHeader::getBackMesTimes, QuantityConstant.MAX_MES_TRY_TIMES);
+        List<TaskHeader> taskHeaderList = taskHeaderService.list(taskHeaderLambdaQueryWrapper);
+        for (TaskHeader taskHeader : taskHeaderList) {
+            try {
+                taskHeaderService.addMesTryTimes(taskHeader);
+                mesService.backMesReceipt(taskHeader);
+            } catch (Exception e) {
+                e.printStackTrace();
             }
+        }
-            taskHeaderLambdaQueryWrapper = Wrappers.lambdaQuery();
-            taskHeaderLambdaQueryWrapper.eq(TaskHeader::getStatus, QuantityConstant.TASK_STATUS_COMPLETED).ne(TaskHeader::getBackMes, QuantityConstant.MES_SEND_OK)
-                .in(TaskHeader::getTaskType, QuantityConstant.TASK_TYPE_WHOLESHIPMENT, QuantityConstant.TASK_TYPE_SORTINGSHIPMENT)
-                .le(TaskHeader::getBackMesTimes, QuantityConstant.MAX_MES_TRY_TIMES);
-            taskHeaderList = taskHeaderService.list(taskHeaderLambdaQueryWrapper);
-            for (TaskHeader taskHeader : taskHeaderList) {
-                try {
-                    taskHeaderService.addMesTryTimes(taskHeader);
-                    mesService.backMesShipment(taskHeader);
-                } catch (Exception e) {
-                    e.printStackTrace();
-                }
+        taskHeaderLambdaQueryWrapper = Wrappers.lambdaQuery();
+        taskHeaderLambdaQueryWrapper.eq(TaskHeader::getStatus, QuantityConstant.TASK_STATUS_COMPLETED).ne(TaskHeader::getBackMes, QuantityConstant.MES_SEND_OK)
+            .in(TaskHeader::getTaskType, QuantityConstant.TASK_TYPE_WHOLESHIPMENT, QuantityConstant.TASK_TYPE_SORTINGSHIPMENT)
+            .le(TaskHeader::getBackMesTimes, QuantityConstant.MAX_MES_TRY_TIMES);
+        taskHeaderList = taskHeaderService.list(taskHeaderLambdaQueryWrapper);
+        for (TaskHeader taskHeader : taskHeaderList) {
+            try {
+                taskHeaderService.addMesTryTimes(taskHeader);
+                mesService.backMesShipment(taskHeader);
+            } catch (Exception e) {
+                e.printStackTrace();
             }
-        } catch (Exception e) {
-            e.printStackTrace();
-        } finally {
-            runningTaskMap.put(taskKey, false);
         }
     }
 }
@@ -11,9 +11,11 @@ import org.jeecg.modules.wms.api.mes.servuce.IMesService;
 import org.jeecg.modules.wms.task.taskHeader.entity.TaskHeader;
 import org.jeecg.modules.wms.task.taskHeader.service.ITaskHeaderService;
 import org.jeecg.utils.constant.QuantityConstant;
+import org.quartz.DisallowConcurrentExecution;
 import org.quartz.Job;
 import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
+import org.quartz.PersistJobDataAfterExecution;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
@@ -24,14 +26,15 @@ import lombok.extern.slf4j.Slf4j;
  * @author 游杰
  */
 @Slf4j
+@PersistJobDataAfterExecution
+@DisallowConcurrentExecution
 public class MesTryTask implements Job {
     @Resource
     private IMesService mesService;
+
     @Resource
     private ITaskHeaderService taskHeaderService;
-    // 并发控制
-    Map<String, Boolean> runningTaskMap = new HashMap<>();
     /**
      * 找到完成状态的，没有回传成功的所有任务，然后回传信息给MES
@@ -40,43 +43,32 @@ public class MesTryTask implements Job {
      */
     @Override
     public void execute(JobExecutionContext context) throws JobExecutionException {
-        String taskKey = "MesTrYTask";
-        if (MapUtils.getBoolean(runningTaskMap, taskKey, false)) {
-            return;
-        }
-        try {
-            runningTaskMap.put(taskKey, true);
-            LambdaQueryWrapper<TaskHeader> taskHeaderLambdaQueryWrapper = Wrappers.lambdaQuery();
-            taskHeaderLambdaQueryWrapper.eq(TaskHeader::getStatus, QuantityConstant.TASK_STATUS_COMPLETED).ne(TaskHeader::getBackMes, QuantityConstant.MES_SEND_OK)
-                .in(TaskHeader::getTaskType, QuantityConstant.TASK_TYPE_WHOLERECEIPT, QuantityConstant.TASK_TYPE_SUPPLEMENTRECEIPT)
-                .gt(TaskHeader::getBackMesTimes, QuantityConstant.MAX_MES_TRY_TIMES);
-            List<TaskHeader> taskHeaderList = taskHeaderService.list(taskHeaderLambdaQueryWrapper);
-            for (TaskHeader taskHeader : taskHeaderList) {
-                try {
-                    taskHeaderService.addMesTryTimes(taskHeader);
-                    mesService.backMesReceipt(taskHeader);
-                } catch (Exception e) {
-                    e.printStackTrace();
-                }
+        LambdaQueryWrapper<TaskHeader> taskHeaderLambdaQueryWrapper = Wrappers.lambdaQuery();
+        taskHeaderLambdaQueryWrapper.eq(TaskHeader::getStatus, QuantityConstant.TASK_STATUS_COMPLETED).ne(TaskHeader::getBackMes, QuantityConstant.MES_SEND_OK)
+            .in(TaskHeader::getTaskType, QuantityConstant.TASK_TYPE_WHOLERECEIPT, QuantityConstant.TASK_TYPE_SUPPLEMENTRECEIPT)
+            .gt(TaskHeader::getBackMesTimes, QuantityConstant.MAX_MES_TRY_TIMES);
+        List<TaskHeader> taskHeaderList = taskHeaderService.list(taskHeaderLambdaQueryWrapper);
+        for (TaskHeader taskHeader : taskHeaderList) {
+            try {
+                taskHeaderService.addMesTryTimes(taskHeader);
+                mesService.backMesReceipt(taskHeader);
+            } catch (Exception e) {
+                e.printStackTrace();
             }
+        }
-            taskHeaderLambdaQueryWrapper = Wrappers.lambdaQuery();
-            taskHeaderLambdaQueryWrapper.eq(TaskHeader::getStatus, QuantityConstant.TASK_STATUS_COMPLETED).ne(TaskHeader::getBackMes, QuantityConstant.MES_SEND_OK)
-                .in(TaskHeader::getTaskType, QuantityConstant.TASK_TYPE_WHOLESHIPMENT, QuantityConstant.TASK_TYPE_SORTINGSHIPMENT)
-                .gt(TaskHeader::getBackMesTimes, QuantityConstant.MAX_MES_TRY_TIMES);
-            taskHeaderList = taskHeaderService.list(taskHeaderLambdaQueryWrapper);
-            for (TaskHeader taskHeader : taskHeaderList) {
-                try {
-                    taskHeaderService.addMesTryTimes(taskHeader);
-                    mesService.backMesShipment(taskHeader);
-                } catch (Exception e) {
-                    e.printStackTrace();
-                }
+        taskHeaderLambdaQueryWrapper = Wrappers.lambdaQuery();
+        taskHeaderLambdaQueryWrapper.eq(TaskHeader::getStatus, QuantityConstant.TASK_STATUS_COMPLETED).ne(TaskHeader::getBackMes, QuantityConstant.MES_SEND_OK)
+            .in(TaskHeader::getTaskType, QuantityConstant.TASK_TYPE_WHOLESHIPMENT, QuantityConstant.TASK_TYPE_SORTINGSHIPMENT)
+            .gt(TaskHeader::getBackMesTimes, QuantityConstant.MAX_MES_TRY_TIMES);
+        taskHeaderList = taskHeaderService.list(taskHeaderLambdaQueryWrapper);
+        for (TaskHeader taskHeader : taskHeaderList) {
+            try {
+                taskHeaderService.addMesTryTimes(taskHeader);
+                mesService.backMesShipment(taskHeader);
+            } catch (Exception e) {
+                e.printStackTrace();
             }
-        } catch (Exception e) {
-            e.printStackTrace();
-        } finally {
-            runningTaskMap.put(taskKey, false);
         }
     }
 }
@@ -12,9 +12,11 @@ import org.jeecg.modules.wms.framework.service.IHuahengMultiHandlerService;
 import org.jeecg.modules.wms.task.taskHeader.entity.TaskHeader;
 import org.jeecg.modules.wms.task.taskHeader.service.ITaskHeaderService;
 import org.jeecg.utils.constant.QuantityConstant;
+import org.quartz.DisallowConcurrentExecution;
 import org.quartz.Job;
 import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
+import org.quartz.PersistJobDataAfterExecution;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
@@ -27,38 +29,28 @@ import lombok.extern.slf4j.Slf4j;
  */
 @Slf4j
+@PersistJobDataAfterExecution
+@DisallowConcurrentExecution
 public class WcsTask implements Job {
-    // 并发控制
-    Map<String, Boolean> runningTaskMap = new HashMap<>();
     @Resource
     private ITaskHeaderService taskHeaderService;
+
     @Resource
     private IHuahengMultiHandlerService huahengMultiHandlerService;
     @Override
     public void execute(JobExecutionContext context) throws JobExecutionException {
-        log.info(String.format(" Jeecg-Boot 执行任务!  时间:" + DateUtils.getTimestamp()));
-        String taskKey = "RyTask";
-        if (MapUtils.getBoolean(runningTaskMap, taskKey, false)) {
-            return;
-        }
-        try {
-            runningTaskMap.put(taskKey, true);
-            LambdaQueryWrapper<TaskHeader> taskHeaderLambdaQueryWrapper = Wrappers.lambdaQuery();
-            taskHeaderLambdaQueryWrapper.eq(TaskHeader::getStatus, QuantityConstant.TASK_STATUS_BUILD).eq(TaskHeader::getAgvTaskId, 0);
-            List<TaskHeader> taskHeaderList = taskHeaderService.list(taskHeaderLambdaQueryWrapper);
-            for (TaskHeader taskHeader : taskHeaderList) {
-                try {
-                    huahengMultiHandlerService.sendTaskToWcs(taskHeader);
-                } catch (Exception e) {
-                    e.printStackTrace();
-                }
+        log.info(String.format(" WcsTask 执行任务! 时间:" + DateUtils.getTimestamp()));
+        LambdaQueryWrapper<TaskHeader> taskHeaderLambdaQueryWrapper = Wrappers.lambdaQuery();
+        taskHeaderLambdaQueryWrapper.eq(TaskHeader::getStatus, QuantityConstant.TASK_STATUS_BUILD).eq(TaskHeader::getAgvTaskId, 0);
+        List<TaskHeader> taskHeaderList = taskHeaderService.list(taskHeaderLambdaQueryWrapper);
+        for (TaskHeader taskHeader : taskHeaderList) {
+            try {
+                huahengMultiHandlerService.sendTaskToWcs(taskHeader);
+            } catch (Exception e) {
+                e.printStackTrace();
             }
-        } catch (Exception e) {
-            e.printStackTrace();
-        } finally {
-            runningTaskMap.put(taskKey, false);
         }
     }
 }