【采集调度器-后端TCP API】采集调度器下发任务命令 API

collector-scheduling-management/src/main/java/com/docus/server/common/SchedulerTask.java

@@ -11,8 +11,10 @@ import com.docus.server.dto.scheduling.management.schcollector.task.ReportTaskTw
 import com.docus.server.dto.scheduling.management.schcollector.task.SchCollectorTaskDTO;
 import com.docus.server.dto.scheduling.management.schterminator.NettyTerminatorDTO;
 import com.docus.server.entity.scheduling.management.SchCollectRecord;
+import com.docus.server.entity.scheduling.management.SchCollectRecordRetryLog;
 import com.docus.server.enums.BusyStateEnum;
 import com.docus.server.enums.RetryTaskEnum;
+import com.docus.server.service.ISchCollectRecordRetryLogService;
 import com.docus.server.service.ISchCollectRecordService;
 import com.docus.server.service.ISchTerminatorService;
 import com.google.common.collect.Lists;
@@ -27,6 +29,9 @@ import org.springframework.util.CollectionUtils;
 import javax.annotation.Resource;
 import java.util.Date;
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
 
 @Component
 @Slf4j
@@ -38,8 +43,12 @@ public class SchedulerTask {
     @Resource
     private ISchCollectRecordService iSchCollectRecordService;
     @Resource
+    private ISchCollectRecordRetryLogService iSchCollectRecordRetryLogService;
+    @Resource
     private ISchTerminatorService iSchTerminatorService;
 
+    private BlockingQueue<ReportDownTwoDTO> retryTaskQueue = new LinkedBlockingQueue<>();
+
     //定时任务
     // 5 * * * * ? 在每分钟的5秒执行
     @Scheduled(cron = "0/1 * * * * ?")
@@ -186,12 +195,25 @@ public class SchedulerTask {
     private void tcpToClient(NettyTerminatorDTO terminal, ReportDownTwoDTO reportDownTwoDTO) {
         Channel channel = channelRepository.get(terminal.getTerminatorIp());
 
-        SchCollectRecord schCollectRecord = iSchCollectRecordService.saveOrUpdateRecord(terminal, reportDownTwoDTO);
+        Map<String, Object> params = reportDownTwoDTO.getParams();
 
         SchCollectorTaskDTO messageContent = new SchCollectorTaskDTO();
-        messageContent.setCollectorRecordId(schCollectRecord.getId());
+
-        messageContent.setIsRetry(String.valueOf(RetryTaskEnum.NO_RETRY_TASK.getValue()));
+        if (CollectionUtils.isEmpty(params)) {
-        messageContent.setTaskInfo(reportDownTwoDTO);
+            //不是重试任务
+            SchCollectRecord schCollectRecord = iSchCollectRecordService.saveOrUpdateRecord(terminal, reportDownTwoDTO);
+
+            messageContent.setCollectorRecordId(schCollectRecord.getId());
+            messageContent.setIsRetry(String.valueOf(RetryTaskEnum.NO_RETRY_TASK.getValue()));
+            messageContent.setTaskInfo(reportDownTwoDTO);
+        } else {
+            //重试任务
+            SchCollectRecordRetryLog schCollectRecordRetryLog = iSchCollectRecordRetryLogService.saveOrUpdateRecordRetryLog(terminal, reportDownTwoDTO);
+
+            messageContent.setCollectorRecordId(schCollectRecordRetryLog.getId());
+            messageContent.setIsRetry(String.valueOf(RetryTaskEnum.RETRY_TASK.getValue()));
+            messageContent.setTaskInfo(reportDownTwoDTO);
+        }
 
         CommMsg commMsg = CommMsg.builder()
             .messageType(MsgConstants.SCH_DISTRIBUTE_TASKS)
@@ -205,8 +227,14 @@ public class SchedulerTask {
         }
     }
 
+
+    public void addRetryTask(ReportDownTwoDTO reportDownTwoDTO) {
+        this.retryTaskQueue.add(reportDownTwoDTO);
+    }
+
+
     //根据有效终端，一次获取一批任务，例如10个终端，获取10个不同类型任务
-    private List<ReportDownTwoDTO> getTask(int size) {
+    public List<ReportDownTwoDTO> getTask(int size) throws InterruptedException {
 //        return taskDistributeApi.getTask(String.valueOf(size));
 
         //mock
@@ -303,8 +331,16 @@ public class SchedulerTask {
             "}";
         ReportDownTwoDTO reportDownTwoDTO3 = JSON.fromJSON(collectorId3, ReportDownTwoDTO.class);
 
+        List<ReportDownTwoDTO> allTaskList = Lists.newArrayList(reportDownTwoDTO1, reportDownTwoDTO2, reportDownTwoDTO3);
+
+        if (!CollectionUtils.isEmpty(this.retryTaskQueue)) {
+            ReportDownTwoDTO retryTask = (ReportDownTwoDTO) this.retryTaskQueue.take();
+            //重试任务
+            allTaskList.add(retryTask);
+        }
+
+        return allTaskList;
 
-        return Lists.newArrayList(reportDownTwoDTO1, reportDownTwoDTO2, reportDownTwoDTO3);
     }
 
     //根据采集器id类型，一次获取一批采集器类型任务

collector-scheduling-management/src/main/java/com/docus/server/controller/SchCollectRecordController.java

@@ -1,22 +1,22 @@
 package com.docus.server.controller;
 
+import com.docus.core.util.ParamsUtils;
 import com.docus.core.util.json.JSON;
 import com.docus.infrastructure.web.request.SearchDTO;
 import com.docus.infrastructure.web.response.PageResult;
 import com.docus.log.annotation.TrackGroup;
 import com.docus.server.api.scheduling.management.SchCollectRecordApi;
-import com.docus.server.common.MsgConstants;
+import com.docus.server.common.SchedulerTask;
 import com.docus.server.common.process.ChannelProcessor;
-import com.docus.server.dto.scheduling.management.schcollector.task.SchCollectorTaskDTO;
+import com.docus.server.dto.scheduling.management.schcollector.task.ReportDownTwoDTO;
 import com.docus.server.dto.scheduling.management.schcollectrecord.AddSchCollectRecordDTO;
 import com.docus.server.dto.scheduling.management.schcollectrecord.DeleteSchCollectRecordDTO;
 import com.docus.server.dto.scheduling.management.schcollectrecord.EditSchCollectRecordDTO;
-import com.docus.server.dto.scheduling.management.schterminator.CommMsgDTO;
+import com.docus.server.dto.scheduling.management.schcollectrecord.RetrySchCollectRecordDTO;
 import com.docus.server.entity.scheduling.management.SchCollectRecord;
-import com.docus.server.service.ICommMsgService;
+import com.docus.server.enums.RetryTaskEnum;
 import com.docus.server.service.ISchCollectRecordService;
 import com.docus.server.vo.scheduling.management.schcollectrecord.SchCollectRecordVO;
-import com.fasterxml.jackson.core.type.TypeReference;
 import org.springframework.web.bind.annotation.RestController;
 
 import javax.annotation.Resource;
@@ -32,7 +32,7 @@ public class SchCollectRecordController implements SchCollectRecordApi {
     @Resource
     private ISchCollectRecordService iSchCollectRecordService;
     @Resource
-    private ICommMsgService iCommMsgService;
+    private SchedulerTask schedulerTask;
 
     /**
      * 按主键查询
@@ -42,44 +42,6 @@ public class SchCollectRecordController implements SchCollectRecordApi {
      */
     @Override
     public SchCollectRecordVO findById(String id) {
-        String s = "{\n" +
-            "\t\"taskInfo\": {\n" +
-            "\t\t\"createTime\": \"2022-12-03 12:39:30\",\n" +
-            "\t\t\"hospitals\": [{\n" +
-            "\t\t\t\"admissDate\": \"2023-12-31 01:01:01\",\n" +
-            "\t\t\t\"admissId\": \"amid_999901\",\n" +
-            "\t\t\t\"admissTimes\": 1,\n" +
-            "\t\t\t\"disDate\": \"2023-12-31 01:01:01\",\n" +
-            "\t\t\t\"disDeptName\": \"22222\",\n" +
-            "\t\t\t\"patientId\": \"758878610105573376\"\n" +
-            "\t\t}],\n" +
-            "\t\t\"jzh\": \"jzh_999901\",\n" +
-            "\t\t\"patient\": {\n" +
-            "\t\t\t\"inpatientNo\": \"999901\",\n" +
-            "\t\t\t\"name\": \"ceshi\",\n" +
-            "\t\t\t\"patientId\": \"758878610105573376\"\n" +
-            "\t\t},\n" +
-            "\t\t\"patientId\": \"758878610105573376\",\n" +
-            "\t\t\"recordType\": \"1\",\n" +
-            "\t\t\"tasks\": [{\n" +
-            "\t\t\t\"collectorId\": \"1\",\n" +
-            "\t\t\t\"patientId\": \"758878610105573376\",\n" +
-            "\t\t\t\"taskId\": \"834292710565826560\"\n" +
-            "\t\t}]\n" +
-            "\t},\n" +
-            "\t\"collectorRecordId\": \"123\",\n" +
-            "\t\"isRetry\": 0\n" +
-            "\n" +
-            "}";
-        SchCollectorTaskDTO schCollectorTaskDTO = JSON.fromJSONWithGeneric(s, new TypeReference<SchCollectorTaskDTO>() {
-        });
-
-        CommMsgDTO commMsgDTO = CommMsgDTO.builder()
-            .content(JSON.toJSON(schCollectorTaskDTO))
-            .messageType(MsgConstants.SCH_DISTRIBUTE_TASKS)
-            .build();
-
-        iCommMsgService.clientsCommand(commMsgDTO);
         return iSchCollectRecordService.findById(id);
     }
 
@@ -127,4 +89,15 @@ public class SchCollectRecordController implements SchCollectRecordApi {
     public int delete(DeleteSchCollectRecordDTO deleteSchCollectRecordDTO) {
         return iSchCollectRecordService.delete(deleteSchCollectRecordDTO);
     }
+
+    @Override
+    public void retryTask(RetrySchCollectRecordDTO retrySchCollectRecordDTO) {
+        ReportDownTwoDTO report = JSON.fromJSON(retrySchCollectRecordDTO.getTaskOriginJson(), ReportDownTwoDTO.class);
+
+        report.setParams(ParamsUtils
+            .addParam("collectRecordId", retrySchCollectRecordDTO.getId())
+            .addParam("isRetryTask", RetryTaskEnum.RETRY_TASK.getValue())
+            .param());
+        schedulerTask.addRetryTask(report);
+    }
 }

collector-scheduling-management/src/main/java/com/docus/server/controller/SchCollectRecordRetryLogController.java

@@ -6,6 +6,7 @@ import com.docus.server.api.scheduling.management.SchCollectRecordRetryLogApi;
 import com.docus.server.dto.scheduling.management.schcollectrecordretrylog.AddSchCollectRecordRetryLogDTO;
 import com.docus.server.dto.scheduling.management.schcollectrecordretrylog.DeleteSchCollectRecordRetryLogDTO;
 import com.docus.server.dto.scheduling.management.schcollectrecordretrylog.EditSchCollectRecordRetryLogDTO;
+import com.docus.server.entity.scheduling.management.SchCollectRecordRetryLog;
 import com.docus.server.service.ISchCollectRecordRetryLogService;
 import com.docus.server.vo.scheduling.management.schcollectrecordretrylog.SchCollectRecordRetryLogVO;
 import org.springframework.web.bind.annotation.RestController;
@@ -52,7 +53,7 @@ public class SchCollectRecordRetryLogController implements SchCollectRecordRetry
      * @return 成功或失败
      */
     @Override
-    public boolean add(AddSchCollectRecordRetryLogDTO addSchCollectRecordRetryLogDTO) {
+    public SchCollectRecordRetryLog add(AddSchCollectRecordRetryLogDTO addSchCollectRecordRetryLogDTO) {
         return iSchCollectRecordRetryLogService.add(addSchCollectRecordRetryLogDTO);
     }
 

collector-scheduling-management/src/main/java/com/docus/server/service/ISchCollectRecordRetryLogService.java

@@ -2,9 +2,12 @@ package com.docus.server.service;
 
 import com.docus.infrastructure.web.request.SearchDTO;
 import com.docus.infrastructure.web.response.PageResult;
+import com.docus.server.dto.scheduling.management.schcollector.task.ReportDownTwoDTO;
 import com.docus.server.dto.scheduling.management.schcollectrecordretrylog.AddSchCollectRecordRetryLogDTO;
 import com.docus.server.dto.scheduling.management.schcollectrecordretrylog.DeleteSchCollectRecordRetryLogDTO;
 import com.docus.server.dto.scheduling.management.schcollectrecordretrylog.EditSchCollectRecordRetryLogDTO;
+import com.docus.server.dto.scheduling.management.schterminator.NettyTerminatorDTO;
+import com.docus.server.entity.scheduling.management.SchCollectRecordRetryLog;
 import com.docus.server.vo.scheduling.management.schcollectrecordretrylog.SchCollectRecordRetryLogVO;
 
 /**
@@ -28,7 +31,7 @@ public interface ISchCollectRecordRetryLogService {
      * @param addSchCollectRecordRetryLogDTO 新增参数
      * @return 成功或失败
      */
-    boolean add(AddSchCollectRecordRetryLogDTO addSchCollectRecordRetryLogDTO);
+    SchCollectRecordRetryLog add(AddSchCollectRecordRetryLogDTO addSchCollectRecordRetryLogDTO);
 
     /**
      * 编辑
@@ -53,4 +56,6 @@ public interface ISchCollectRecordRetryLogService {
      * @return 分页列表
      */
     PageResult<SchCollectRecordRetryLogVO> search(SearchDTO searchDTO);
+
+    SchCollectRecordRetryLog saveOrUpdateRecordRetryLog(NettyTerminatorDTO terminal, ReportDownTwoDTO reportDownTwoDTO);
 }

collector-scheduling-management/src/main/java/com/docus/server/service/impl/SchCollectRecordRetryLogServiceImpl.java

@@ -1,20 +1,27 @@
 package com.docus.server.service.impl;
 
+import com.docus.core.util.DateUtil;
 import com.docus.core.util.ListUtils;
+import com.docus.core.util.json.JSON;
 import com.docus.core.util.property.Setters;
+import com.docus.infrastructure.redis.service.IdService;
 import com.docus.infrastructure.web.request.SearchDTO;
 import com.docus.infrastructure.web.response.PageResult;
 import com.docus.server.convert.SchCollectRecordRetryLogConvert;
+import com.docus.server.dto.scheduling.management.schcollector.task.ReportDownTwoDTO;
 import com.docus.server.dto.scheduling.management.schcollectrecordretrylog.AddSchCollectRecordRetryLogDTO;
 import com.docus.server.dto.scheduling.management.schcollectrecordretrylog.DeleteSchCollectRecordRetryLogDTO;
 import com.docus.server.dto.scheduling.management.schcollectrecordretrylog.EditSchCollectRecordRetryLogDTO;
+import com.docus.server.dto.scheduling.management.schterminator.NettyTerminatorDTO;
 import com.docus.server.entity.scheduling.management.SchCollectRecordRetryLog;
 import com.docus.server.entity.scheduling.management.SchSystemParams;
+import com.docus.server.enums.SubStateEnum;
 import com.docus.server.infrastructure.dao.ISchCollectRecordRetryLogDao;
 import com.docus.server.infrastructure.dao.ISchSystemParamsDao;
 import com.docus.server.service.ISchCollectRecordRetryLogService;
 import com.docus.server.vo.scheduling.management.schcollectrecordretrylog.SchCollectRecordRetryLogVO;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.CollectionUtils;
 
 import javax.annotation.Resource;
@@ -33,6 +40,8 @@ public class SchCollectRecordRetryLogServiceImpl implements ISchCollectRecordRet
     private ISchCollectRecordRetryLogDao iSchCollectRecordRetryLogDao;
     @Resource
     private ISchSystemParamsDao iSchSystemParamsDao;
+    @Resource
+    private IdService idService;
 
     /**
      * 按主键查询
@@ -72,6 +81,27 @@ public class SchCollectRecordRetryLogServiceImpl implements ISchCollectRecordRet
         return result;
     }
 
+    @Transactional(rollbackFor = Exception.class)
+    @Override
+    public SchCollectRecordRetryLog saveOrUpdateRecordRetryLog(NettyTerminatorDTO terminal, ReportDownTwoDTO messageContent) {
+        //新增采集记录表
+        AddSchCollectRecordRetryLogDTO retryLogDTO = new AddSchCollectRecordRetryLogDTO();
+        retryLogDTO.setId(idService.getDateSeq());
+        retryLogDTO.setCollectRecordId(Long.valueOf(String.valueOf(messageContent.getParams().get("collectRecordId"))));
+        retryLogDTO.setCollectorId(Long.valueOf(messageContent.getTasks().get(0).getCollectorId()));
+        retryLogDTO.setTerminatorId(terminal.getId());
+        retryLogDTO.setTaskId(messageContent.getTasks().get(0).getTaskId());
+        SchSystemParams params = iSchSystemParamsDao.findOneBy("paramValue", messageContent.getTasks().get(0).getCollectorId());
+        retryLogDTO.setTaskName(String.format("%s%s%s", messageContent.getPatient().getName(), params.getParamName(), "采集"));
+        retryLogDTO.setTaskMemo(String.format("%s%s%s", messageContent.getPatient().getName(), params.getParamName(), "采集"));
+        retryLogDTO.setStartTime(DateUtil.now());
+        retryLogDTO.setTaskDetailInfo(String.format("病案号：%s，姓名：%s，采集类型：%s", messageContent.getPatientId(), messageContent.getPatient().getName(), params.getParamName()));
+        retryLogDTO.setSubTaskExecState(SubStateEnum.RECEIVE);
+        retryLogDTO.setTaskOriginJson(JSON.toJSON(messageContent));
+
+        return this.add(retryLogDTO);
+    }
+
     /**
      * 新增
      *
@@ -79,9 +109,10 @@ public class SchCollectRecordRetryLogServiceImpl implements ISchCollectRecordRet
      * @return 成功或失败
      */
     @Override
-    public boolean add(AddSchCollectRecordRetryLogDTO addSchCollectRecordRetryLogDTO) {
+    public SchCollectRecordRetryLog add(AddSchCollectRecordRetryLogDTO addSchCollectRecordRetryLogDTO) {
         SchCollectRecordRetryLog schCollectRecordRetryLog = SchCollectRecordRetryLogConvert.INSTANCE.convertDO(addSchCollectRecordRetryLogDTO);
-        return iSchCollectRecordRetryLogDao.add(schCollectRecordRetryLog);
+        iSchCollectRecordRetryLogDao.add(schCollectRecordRetryLog);
+        return schCollectRecordRetryLog;
     }
 
     /**

docus-client-interface/src/main/java/com/docus/server/api/scheduling.management/SchCollectRecordApi.java

@@ -5,6 +5,7 @@ import com.docus.infrastructure.web.response.PageResult;
 import com.docus.server.dto.scheduling.management.schcollectrecord.AddSchCollectRecordDTO;
 import com.docus.server.dto.scheduling.management.schcollectrecord.DeleteSchCollectRecordDTO;
 import com.docus.server.dto.scheduling.management.schcollectrecord.EditSchCollectRecordDTO;
+import com.docus.server.dto.scheduling.management.schcollectrecord.RetrySchCollectRecordDTO;
 import com.docus.server.entity.scheduling.management.SchCollectRecord;
 import com.docus.server.vo.scheduling.management.schcollectrecord.SchCollectRecordVO;
 import io.swagger.annotations.Api;
@@ -79,4 +80,15 @@ public interface SchCollectRecordApi {
     @ApiOperation("批量删除")
     @DeleteMapping("/delete")
     int delete(@RequestBody DeleteSchCollectRecordDTO deleteSchCollectRecordDTO);
+
+    /**
+     * 任务重试
+     *
+     * @param addSchCollectRecordDTO 新增参数
+     */
+    @ApiOperation("任务重试")
+    @PostMapping("/retryTask")
+    void retryTask(@RequestBody RetrySchCollectRecordDTO addSchCollectRecordDTO);
+
+
 }

docus-client-interface/src/main/java/com/docus/server/api/scheduling.management/SchCollectRecordRetryLogApi.java

@@ -5,6 +5,7 @@ import com.docus.infrastructure.web.response.PageResult;
 import com.docus.server.dto.scheduling.management.schcollectrecordretrylog.AddSchCollectRecordRetryLogDTO;
 import com.docus.server.dto.scheduling.management.schcollectrecordretrylog.DeleteSchCollectRecordRetryLogDTO;
 import com.docus.server.dto.scheduling.management.schcollectrecordretrylog.EditSchCollectRecordRetryLogDTO;
+import com.docus.server.entity.scheduling.management.SchCollectRecordRetryLog;
 import com.docus.server.vo.scheduling.management.schcollectrecordretrylog.SchCollectRecordRetryLogVO;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
@@ -57,7 +58,7 @@ public interface SchCollectRecordRetryLogApi {
      */
     @ApiOperation("新增")
     @PostMapping("/add")
-    boolean add(@RequestBody AddSchCollectRecordRetryLogDTO addSchCollectRecordRetryLogDTO);
+    SchCollectRecordRetryLog add(@RequestBody AddSchCollectRecordRetryLogDTO addSchCollectRecordRetryLogDTO);
 
     /**
      * 编辑

docus-client-interface/src/main/java/com/docus/server/dto/scheduling.management/schcollector/task/ReportDownTwoDTO.java

@@ -9,6 +9,7 @@ import lombok.NoArgsConstructor;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 
 /**
  * 下发任务
@@ -20,18 +21,27 @@ import java.util.List;
 @ApiModel(value = "ReportDownTwoDTO", description = "下发任务")
 public class ReportDownTwoDTO {
 
+    @ApiModelProperty(value = "扩展参数")
+    private Map<String, Object> params;
+
     @ApiModelProperty(value = "档案编号")
     private String patientId;
+
     @ApiModelProperty(value = "记账号")
     private String jzh;
+
     @ApiModelProperty(value = "档案类型")
     private String recordType;
+
     @ApiModelProperty(value = "档案时间")
     private String createTime;
+
     @ApiModelProperty(value = "患者信息")
     private ReportPatientTwoDTO patient = new ReportPatientTwoDTO();
+
     @ApiModelProperty(value = "住院信息")
     private List<ReportHospitalTwoDTO> hospitals = new ArrayList<>();
+
     @ApiModelProperty(value = "任务信息")
     private List<ReportTaskTwoDTO> tasks = new ArrayList<>();
 

docus-client-interface/src/main/java/com/docus/server/dto/scheduling.management/schcollectrecord/RetrySchCollectRecordDTO.java

@@ -0,0 +1,26 @@
+package com.docus.server.dto.scheduling.management.schcollectrecord;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * 采集记录表 AddDTO
+ *
+ * @author AutoGenerator
+ * @since 2023-07-15
+ */
+@Data
+@ApiModel(value = "RetrySchCollectRecordDTO", description = "采集记录表")
+public class RetrySchCollectRecordDTO implements Serializable {
+
+    @ApiModelProperty(value = "主键")
+    private Long id;
+
+    @ApiModelProperty(value = "redis队列任务原始报文")
+    private String taskOriginJson;
+
+
+}