-

epmet-user/epmet-user-server/src/main/java/com/epmet/dao/IcSyncJobDao.java

@@ -17,6 +17,5 @@ import java.util.List;
 public interface IcSyncJobDao extends BaseDao<IcSyncJobEntity> {
 
     List<IcSyncJobEntity> selectExecutableJobList(@Param("jobType") String jobType,
-                                                  @Param("operationStatus") String operationStatus,
                                                   @Param("itemCount") int itemCount);
 }

epmet-user/epmet-user-server/src/main/java/com/epmet/processor/YanTaiNatSyncProcessor.java

@@ -55,10 +55,10 @@ public class YanTaiNatSyncProcessor {
     public void scanJobs() {
         log.info("【异步数据更新】开始同步任务");
 
-        String dataSyncEnable = redisUtils.getString("data:sync:enable");
-        if (StringUtils.isEmpty(dataSyncEnable)) {
-            return;
-        }
+        //String dataSyncEnable = redisUtils.getString("data:sync:enable");
+        //if (StringUtils.isEmpty(dataSyncEnable)) {
+        //    return;
+        //}
 
         LambdaQueryWrapper<IcSyncJobEntity> executingListQuery = new LambdaQueryWrapper<>();
         executingListQuery.eq(IcSyncJobEntity::getOperationStatus, EpidemicConstant.OPERATION_STATUS_PROCESSING);
@@ -79,7 +79,6 @@ public class YanTaiNatSyncProcessor {
             // 查询可执行的任务列表，并且异步执行
             List<IcSyncJobEntity> icSyncJobToExec = icSyncJobDao.selectExecutableJobList(
                     EpidemicConstant.JOB_TYPE_NAT,
-                    EpidemicConstant.OPERATION_STATUS_WAITING,
                     leftCount);
 
             if (!CollectionUtils.isEmpty(icSyncJobToExec)) {
@@ -90,10 +89,13 @@ public class YanTaiNatSyncProcessor {
                     executorService.submit(() -> {
                         // 将此任务状态修改为执行中
 
+                        try {
                             dataSyncConfigService.execSyncByJobProcessor(jobEntity);
-
+                        } finally {
                             // 更新任务状态为结束
                             updateJobStatus(jobEntity.getId(), EpidemicConstant.OPERATION_STATUS_FINISH);
+                        }
+
                     });
                 }
             }

epmet-user/epmet-user-server/src/main/java/com/epmet/service/impl/DataSyncConfigServiceImpl.java

@@ -1099,7 +1099,7 @@ public class DataSyncConfigServiceImpl extends BaseServiceImpl<DataSyncConfigDao
                 // 从视图中获取到的核酸采样相关信息
                 String name = (String) resiHscyInfo.get("name");
                 String cardNo = (String) resiHscyInfo.get("card_no");
-                String createTime = (String) resiHscyInfo.get("create_time");
+                Date createTime = (Date) resiHscyInfo.get("create_time");
 
                 // 本地数据库中，居民信息
                 NatUserInfoResultDTO currentResiInfo = idCardAndResiInfo.get(cardNo);
@@ -1112,7 +1112,7 @@ public class DataSyncConfigServiceImpl extends BaseServiceImpl<DataSyncConfigDao
                 e.setUserType(isSync.equals(NumConstant.ONE_STR) ? "manualSync" : "sync");
                 e.setName(StringUtils.isNotBlank(name) ? name : "");
                 e.setIdCard(StringUtils.isNotBlank(cardNo) ? cardNo : "");
-                e.setSampleTime(DateUtils.parseDate(createTime, DateUtils.DATE_TIME_PATTERN));
+                e.setSampleTime(createTime);
                 e.setAgencyId(currentResiInfo.getAgencyId());
                 e.setPids(currentResiInfo.getPids());
                 e.setAttachmentType("");
@@ -1136,9 +1136,9 @@ public class DataSyncConfigServiceImpl extends BaseServiceImpl<DataSyncConfigDao
                 // 从视图中获取到的核酸采样相关信息
                 String name = (String) natResult.get("name");
                 String cardNo = (String) natResult.get("card_no");
-                String testTime = (String) natResult.get("test_time");
+                Date testTime = (Date) natResult.get("test_time");
                 String telephone = (String) natResult.get("telephone");
-                String sampleTime = (String) natResult.get("SAMPLE_TIME");
+                Date sampleTime = (Date) natResult.get("SAMPLE_TIME");
                 String sampleResultPcr = (String) natResult.get("SAMPLE_RESULT_PCR");
                 String samplingOrgPcr = (String) natResult.get("SAMPLING_ORG_PCR");
 
@@ -1153,8 +1153,8 @@ public class DataSyncConfigServiceImpl extends BaseServiceImpl<DataSyncConfigDao
                 e.setName(StringUtils.isNotBlank(name) ? name : "");
                 e.setMobile(StringUtils.isNotBlank(telephone) ? telephone : "");
                 e.setIdCard(StringUtils.isNotBlank(cardNo) ? cardNo : "");
-                e.setNatTime(DateUtils.parseDate(testTime, DateUtils.DATE_TIME_PATTERN));
-                e.setSampleTime(DateUtils.parseDate(sampleTime, DateUtils.DATE_TIME_PATTERN));
+                e.setNatTime(testTime);
+                e.setSampleTime(sampleTime);
                 String resultPcr = sampleResultPcr;
                 //检测结果 转换 我们 0:阴性  1:阳性, 他们 ：1:阳性,2:阴性
                 e.setNatResult(NumConstant.ZERO_STR);
@@ -1187,16 +1187,19 @@ public class DataSyncConfigServiceImpl extends BaseServiceImpl<DataSyncConfigDao
         // 正常状态
         p.setResiStatus("0");
         // 指定组织
-        p.setAgencyId(jobEntity.getOrgId());
+        p.setAgencyId(jobEntity.getOrgIdPath());
 
         List<NatUserInfoResultDTO> resis = null;
         int pageNo = 1;
         int pageSize = 1000;
+        int updatedResiCount = 0;
         do {
             // 分页，一次查询1000居民，循环更新他们的核酸检测信息
             resis = getNatUserInfoFromDb(p, pageNo, pageSize);
             yantaiHsjcByDbView(resis, jobEntity.getCustomerId(), NumConstant.ONE_STR);
             pageNo++;
-        } while (org.springframework.util.CollectionUtils.isEmpty(resis));
+            updatedResiCount += pageSize;
+            log.info("【任务处理器同步数据】组织Id:{},同步类型:{},已完成居民数:{}", jobEntity.getOrgId(), jobEntity.getJobType(), updatedResiCount);
+        } while (!org.springframework.util.CollectionUtils.isEmpty(resis));
     }
 }

epmet-user/epmet-user-server/src/main/resources/mapper/IcSyncJobDao.xml

@@ -20,7 +20,7 @@
                updated_by,
                updated_time
         from ic_sync_job
-        where OPERATION_STATUS = 'wating'
+        where OPERATION_STATUS = 'waiting'
           and JOB_TYPE = #{jobType}
           and DEL_FLAG = 0
         order by CREATED_TIME asc