|
|
|
@ -1,28 +1,37 @@ |
|
|
|
package com.epmet.service.impl; |
|
|
|
import java.util.Date; |
|
|
|
|
|
|
|
import com.alibaba.fastjson.JSON; |
|
|
|
import com.alibaba.fastjson.JSONObject; |
|
|
|
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; |
|
|
|
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; |
|
|
|
import com.epmet.commons.mybatis.entity.BaseEpmetEntity; |
|
|
|
import com.epmet.commons.mybatis.service.impl.BaseServiceImpl; |
|
|
|
import com.epmet.commons.tools.constant.NumConstant; |
|
|
|
import com.epmet.commons.tools.constant.StrConstant; |
|
|
|
import com.epmet.commons.tools.distributedlock.DistributedLock; |
|
|
|
import com.epmet.commons.tools.dto.form.PageFormDTO; |
|
|
|
import com.epmet.commons.tools.dto.result.CustomerStaffInfoCacheResult; |
|
|
|
import com.epmet.commons.tools.dto.result.YtDataSyncResDTO; |
|
|
|
import com.epmet.commons.tools.dto.result.YtHscyResDTO; |
|
|
|
import com.epmet.commons.tools.dto.result.YtHsjcResDTO; |
|
|
|
import com.epmet.commons.tools.enums.GenderEnum; |
|
|
|
import com.epmet.commons.tools.exception.EpmetErrorCode; |
|
|
|
import com.epmet.commons.tools.exception.EpmetException; |
|
|
|
import com.epmet.commons.tools.exception.ExceptionUtils; |
|
|
|
import com.epmet.commons.tools.page.PageData; |
|
|
|
import com.epmet.commons.tools.redis.common.CustomerOrgRedis; |
|
|
|
import com.epmet.commons.tools.redis.common.CustomerStaffRedis; |
|
|
|
import com.epmet.commons.tools.redis.common.bean.AgencyInfoCache; |
|
|
|
import com.epmet.commons.tools.redis.common.bean.GridInfoCache; |
|
|
|
import com.epmet.commons.tools.security.dto.TokenDto; |
|
|
|
import com.epmet.commons.tools.utils.ConvertUtils; |
|
|
|
import com.epmet.commons.tools.utils.DateUtils; |
|
|
|
import com.epmet.commons.tools.utils.YtHsResUtils; |
|
|
|
import com.epmet.constant.EpidemicConstant; |
|
|
|
import com.epmet.dao.DataSyncConfigDao; |
|
|
|
import com.epmet.dao.IcNatDao; |
|
|
|
import com.epmet.dao.IcSyncJobDao; |
|
|
|
import com.epmet.dto.DataSyncConfigDTO; |
|
|
|
import com.epmet.dto.DataSyncRecordDeathDTO; |
|
|
|
import com.epmet.dto.DataSyncRecordDisabilityDTO; |
|
|
|
@ -39,6 +48,7 @@ import lombok.extern.slf4j.Slf4j; |
|
|
|
import org.apache.commons.collections4.CollectionUtils; |
|
|
|
import org.apache.commons.collections4.ListUtils; |
|
|
|
import org.apache.commons.lang3.StringUtils; |
|
|
|
import org.redisson.api.RLock; |
|
|
|
import org.springframework.beans.factory.annotation.Autowired; |
|
|
|
import org.springframework.jdbc.core.JdbcTemplate; |
|
|
|
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; |
|
|
|
@ -49,9 +59,12 @@ import javax.annotation.Resource; |
|
|
|
import java.util.*; |
|
|
|
import java.util.concurrent.CountDownLatch; |
|
|
|
import java.util.concurrent.ExecutorService; |
|
|
|
import java.util.concurrent.TimeUnit; |
|
|
|
import java.util.function.Function; |
|
|
|
import java.util.stream.Collectors; |
|
|
|
|
|
|
|
import static com.epmet.constant.EpidemicConstant.*; |
|
|
|
|
|
|
|
/** |
|
|
|
* 数据更新配置表 |
|
|
|
* |
|
|
|
@ -76,6 +89,12 @@ public class DataSyncConfigServiceImpl extends BaseServiceImpl<DataSyncConfigDao |
|
|
|
private DataSyncRecordDisabilityService dataSyncRecordDisabilityService; |
|
|
|
@Autowired |
|
|
|
private ExecutorService executorService; |
|
|
|
@Autowired |
|
|
|
private IcSyncJobService icSyncJobService; |
|
|
|
@Autowired |
|
|
|
private IcSyncJobDao icSyncJobDao; |
|
|
|
@Autowired |
|
|
|
private DistributedLock distributedLock; |
|
|
|
|
|
|
|
@Resource(name = "yantaiNamedParamLantuJdbcTemplate") |
|
|
|
private NamedParameterJdbcTemplate yantaiNamedParamLantuJdbcTemplate; |
|
|
|
@ -134,7 +153,7 @@ public class DataSyncConfigServiceImpl extends BaseServiceImpl<DataSyncConfigDao |
|
|
|
public PageData list(TokenDto tokenDto, PageFormDTO formDTO) { |
|
|
|
PageData<DataSyncConfigDTO> result = new PageData<>(new ArrayList<>(), NumConstant.ZERO_L); |
|
|
|
PageInfo<DataSyncConfigDTO> pageInfo = PageHelper.startPage(formDTO.getPageNo(), formDTO.getPageSize()) |
|
|
|
.doSelectPageInfo(() -> baseDao.list(tokenDto.getCustomerId(), null)); |
|
|
|
.doSelectPageInfo(() -> baseDao.list(tokenDto.getCustomerId(), null,null)); |
|
|
|
if (CollectionUtils.isNotEmpty(pageInfo.getList())) { |
|
|
|
result.setList(pageInfo.getList()); |
|
|
|
result.setTotal(Integer.parseInt(String.valueOf(pageInfo.getTotal()))); |
|
|
|
@ -189,7 +208,7 @@ public class DataSyncConfigServiceImpl extends BaseServiceImpl<DataSyncConfigDao |
|
|
|
*/ |
|
|
|
@Override |
|
|
|
public void dataSyncForYanTaiTask(DataSyncTaskParam formDTO) { |
|
|
|
List<DataSyncConfigDTO> allConfigList = baseDao.list(formDTO.getCustomerId(), "open"); |
|
|
|
List<DataSyncConfigDTO> allConfigList = baseDao.list(formDTO.getCustomerId(), "open",null); |
|
|
|
if (CollectionUtils.isEmpty(allConfigList)) { |
|
|
|
return; |
|
|
|
} |
|
|
|
@ -225,6 +244,256 @@ public class DataSyncConfigServiceImpl extends BaseServiceImpl<DataSyncConfigDao |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* @Description 死亡信息定时拉取 |
|
|
|
* @param formDTO |
|
|
|
* @Author zxc |
|
|
|
* @Date 2022/11/8 09:01 |
|
|
|
*/ |
|
|
|
@Override |
|
|
|
public void deathInfoScanTask(DataSyncTaskParam formDTO) { |
|
|
|
List<DataSyncConfigDTO> configData = getConfigData(null, EpidemicConstant.DATA_CODE_DEATH); |
|
|
|
if (CollectionUtils.isEmpty(configData)){ |
|
|
|
log.warn("deathInfoScanTask not exists config data,customer is "+formDTO.getCustomerId()); |
|
|
|
return; |
|
|
|
} |
|
|
|
long count = configData.stream().filter(o -> CollectionUtils.isNotEmpty(o.getScopeList())).count(); |
|
|
|
if (count < 1) { |
|
|
|
log.warn("deathInfoScanTask scopeList is null"); |
|
|
|
return; |
|
|
|
} |
|
|
|
int pageNo = NumConstant.ONE; |
|
|
|
int pageSize = NumConstant.ONE_THOUSAND; |
|
|
|
List<NatUserInfoResultDTO> dbResiList = null; |
|
|
|
do { |
|
|
|
for (DataSyncConfigDTO config : configData) { |
|
|
|
// 设置查询数据范围
|
|
|
|
formDTO.setOrgList(config.getScopeList()); |
|
|
|
DataSyncEnum anEnum = DataSyncEnum.getEnum(config.getDataCode()); |
|
|
|
dbResiList = getNatUserInfoFromDb(formDTO, pageNo, pageSize); |
|
|
|
if (CollectionUtils.isEmpty(dbResiList)) { |
|
|
|
continue; |
|
|
|
} |
|
|
|
switch (anEnum) { |
|
|
|
case SI_WANG: |
|
|
|
try { |
|
|
|
//查询正常状态的居民
|
|
|
|
siWang(dbResiList); |
|
|
|
log.info("======siWang信息拉取结束======"); |
|
|
|
} catch (Exception e) { |
|
|
|
log.error("death thread execute exception", e); |
|
|
|
} |
|
|
|
default: |
|
|
|
log.warn("没有要处理的数据"); |
|
|
|
} |
|
|
|
} |
|
|
|
pageNo++; |
|
|
|
} while (dbResiList != null && dbResiList.size() == pageSize); |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* @Description 残疾信息定时拉取 |
|
|
|
* @param formDTO |
|
|
|
* @Author zxc |
|
|
|
* @Date 2022/11/8 09:01 |
|
|
|
*/ |
|
|
|
@Override |
|
|
|
public void disabilityInfoScanTask(DataSyncTaskParam formDTO) { |
|
|
|
List<DataSyncConfigDTO> configData = getConfigData(null, EpidemicConstant.DATA_CODE_DISABILITY); |
|
|
|
if (CollectionUtils.isEmpty(configData)){ |
|
|
|
log.warn("disabilityInfoScanTask not exists config data,customer is "+formDTO.getCustomerId()); |
|
|
|
return; |
|
|
|
} |
|
|
|
long count = configData.stream().filter(o -> CollectionUtils.isNotEmpty(o.getScopeList())).count(); |
|
|
|
if (count < 1) { |
|
|
|
log.warn("disabilityInfoScanTask scopeList is null"); |
|
|
|
return; |
|
|
|
} |
|
|
|
int pageNo = NumConstant.ONE; |
|
|
|
int pageSize = NumConstant.ONE_THOUSAND; |
|
|
|
List<NatUserInfoResultDTO> dbResiList = null; |
|
|
|
do { |
|
|
|
for (DataSyncConfigDTO config : configData) { |
|
|
|
// 设置查询数据范围
|
|
|
|
formDTO.setOrgList(config.getScopeList()); |
|
|
|
formDTO.setCategoryColumn("IS_CJ"); |
|
|
|
DataSyncEnum anEnum = DataSyncEnum.getEnum(config.getDataCode()); |
|
|
|
dbResiList = getNatUserInfoFromDb(formDTO, pageNo, pageSize); |
|
|
|
if (CollectionUtils.isEmpty(dbResiList)) { |
|
|
|
continue; |
|
|
|
} |
|
|
|
switch (anEnum) { |
|
|
|
case CAN_JI: |
|
|
|
try { |
|
|
|
//查询正常状态的居民
|
|
|
|
canJi(dbResiList); |
|
|
|
log.info("======canJi信息拉取结束======"); |
|
|
|
} catch (Exception e) { |
|
|
|
log.error("disability thread execute exception", e); |
|
|
|
} |
|
|
|
break; |
|
|
|
default: |
|
|
|
log.warn("没有要处理的数据"); |
|
|
|
} |
|
|
|
} |
|
|
|
pageNo++; |
|
|
|
} while (dbResiList != null && dbResiList.size() == pageSize); |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* @Description 核酸检测信息定时拉取 |
|
|
|
* @param formDTO |
|
|
|
* @Author zxc |
|
|
|
* @Date 2022/11/8 10:37 |
|
|
|
*/ |
|
|
|
@Override |
|
|
|
public void natInfoScanTask(DataSyncTaskParam formDTO) { |
|
|
|
List<DataSyncConfigDTO> configData = new ArrayList<>(); |
|
|
|
if (StringUtils.isBlank(formDTO.getAgencyId())){ |
|
|
|
configData = getConfigData(null, EpidemicConstant.DATA_CODE_NAT); |
|
|
|
if (CollectionUtils.isEmpty(configData)){ |
|
|
|
log.warn("natInfoScanTask not exists config data,customer is "+formDTO.getCustomerId()); |
|
|
|
return; |
|
|
|
} |
|
|
|
long count = configData.stream().filter(o -> CollectionUtils.isNotEmpty(o.getScopeList())).count(); |
|
|
|
if (count < 1) { |
|
|
|
log.warn("natInfoScanTask scopeList is null"); |
|
|
|
return; |
|
|
|
} |
|
|
|
} |
|
|
|
int pageNo = NumConstant.ONE; |
|
|
|
int pageSize = NumConstant.ONE_THOUSAND; |
|
|
|
List<NatUserInfoResultDTO> dbResiList = null; |
|
|
|
do { |
|
|
|
if (StringUtils.isBlank(formDTO.getAgencyId())){ |
|
|
|
for (DataSyncConfigDTO config : configData) { |
|
|
|
// 设置查询数据范围
|
|
|
|
formDTO.setOrgList(config.getScopeList()); |
|
|
|
DataSyncEnum anEnum = DataSyncEnum.getEnum(config.getDataCode()); |
|
|
|
dbResiList = getNatUserInfoFromDb(formDTO, pageNo, pageSize); |
|
|
|
if (CollectionUtils.isEmpty(dbResiList)) { |
|
|
|
continue; |
|
|
|
} |
|
|
|
switch (anEnum) { |
|
|
|
case HE_SUAN: |
|
|
|
try { |
|
|
|
//查询正常状态的居民
|
|
|
|
yantaiHsjcByDbView(dbResiList, config.getCustomerId(), formDTO.getIsSync()); |
|
|
|
log.info("======核酸检测信息拉取结束======"); |
|
|
|
} catch (Exception e) { |
|
|
|
log.error("nat thread execute exception", e); |
|
|
|
} |
|
|
|
break; |
|
|
|
default: |
|
|
|
log.warn("没有要处理的数据"); |
|
|
|
} |
|
|
|
} |
|
|
|
}else { |
|
|
|
dbResiList = getNatUserInfoFromDb(formDTO, pageNo, pageSize); |
|
|
|
if (CollectionUtils.isEmpty(dbResiList)){ |
|
|
|
return; |
|
|
|
} |
|
|
|
yantaiHsjcByDbView(dbResiList, dbResiList.get(NumConstant.ZERO).getCustomerId(), formDTO.getIsSync()); |
|
|
|
} |
|
|
|
pageNo++; |
|
|
|
} while (dbResiList != null && dbResiList.size() == pageSize); |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void natInfoSyncButton(DataSyncTaskParam formDTO) { |
|
|
|
CustomerStaffInfoCacheResult staffInfo = CustomerStaffRedis.getStaffInfo(formDTO.getCustomerId(), formDTO.getStaffId()); |
|
|
|
if (null == staffInfo){ |
|
|
|
throw new EpmetException("未查询到工作人员信息："+formDTO.getStaffId()); |
|
|
|
} |
|
|
|
AgencyInfoCache agencyInfo = CustomerOrgRedis.getAgencyInfo(formDTO.getAgencyId()); |
|
|
|
if (null == agencyInfo){ |
|
|
|
throw new EpmetException("未查询到组织信息："+formDTO.getAgencyId()); |
|
|
|
} |
|
|
|
LambdaQueryWrapper<IcSyncJobEntity> qw = new LambdaQueryWrapper<>(); |
|
|
|
qw.eq(IcSyncJobEntity::getOrgId,formDTO.getAgencyId()).in(IcSyncJobEntity::getOperationStatus,OPERATION_STATUS_WAITING,OPERATION_STATUS_PROCESSING); |
|
|
|
List<IcSyncJobEntity> icSyncJobEntities = icSyncJobDao.selectList(qw); |
|
|
|
// 当前组织下存在同步任务
|
|
|
|
if (CollectionUtils.isNotEmpty(icSyncJobEntities)){ |
|
|
|
throw new EpmetException(EpmetErrorCode.EXIST_SYNC_JOB_ERROR.getCode()); |
|
|
|
} |
|
|
|
// 不存在新增一条记录
|
|
|
|
IcSyncJobEntity e = new IcSyncJobEntity(); |
|
|
|
e.setCustomerId(formDTO.getCustomerId()); |
|
|
|
e.setOrgId(formDTO.getAgencyId()); |
|
|
|
e.setPid(agencyInfo.getPid()); |
|
|
|
e.setOrgIdPath(StringUtils.isBlank(agencyInfo.getPids()) ? agencyInfo.getId() : agencyInfo.getPids()+":"+agencyInfo.getId()); |
|
|
|
e.setJobType(JOB_TYPE_NAT); |
|
|
|
e.setOperatorId(formDTO.getStaffId()); |
|
|
|
e.setOperationStatus(OPERATION_STATUS_WAITING); |
|
|
|
insertSync(e); |
|
|
|
// 客户下，type=nat 存在两条正在进行中的同步任务
|
|
|
|
LambdaQueryWrapper<IcSyncJobEntity> qw2 = new LambdaQueryWrapper<>(); |
|
|
|
qw2.eq(IcSyncJobEntity::getCustomerId,formDTO.getCustomerId()) |
|
|
|
.eq(IcSyncJobEntity::getOperationStatus,OPERATION_STATUS_PROCESSING) |
|
|
|
.eq(IcSyncJobEntity::getJobType,JOB_TYPE_NAT); |
|
|
|
List<IcSyncJobEntity> processingList = icSyncJobDao.selectList(qw2); |
|
|
|
if (CollectionUtils.isNotEmpty(processingList) && processingList.size() >= NumConstant.TWO){ |
|
|
|
throw new EpmetException(EpmetErrorCode.OTHER_SYNC_JOB_ERROR.getCode()); |
|
|
|
} |
|
|
|
List<IcSyncJobEntity> waitList; |
|
|
|
do { |
|
|
|
LambdaQueryWrapper<IcSyncJobEntity> qw3 = new LambdaQueryWrapper<>(); |
|
|
|
qw3.eq(IcSyncJobEntity::getCustomerId,formDTO.getCustomerId()) |
|
|
|
.eq(IcSyncJobEntity::getOperationStatus,OPERATION_STATUS_WAITING) |
|
|
|
.eq(IcSyncJobEntity::getJobType,JOB_TYPE_NAT) |
|
|
|
.orderByAsc(BaseEpmetEntity::getCreatedTime); |
|
|
|
waitList = icSyncJobDao.selectList(qw3); |
|
|
|
if (CollectionUtils.isNotEmpty(waitList)){ |
|
|
|
for (IcSyncJobEntity entity : waitList) { |
|
|
|
RLock lock = null; |
|
|
|
try { |
|
|
|
lock = distributedLock.getLock(entity.getOrgId() + JOB_TYPE_NAT, 60L, 60L, TimeUnit.SECONDS); |
|
|
|
updateSync(entity.getId(),OPERATION_STATUS_PROCESSING); |
|
|
|
}catch (Exception ex){ |
|
|
|
log.error(ex.getMessage()); |
|
|
|
throw new EpmetException(ex.getMessage()); |
|
|
|
}finally { |
|
|
|
lock.unlock(); |
|
|
|
} |
|
|
|
formDTO.setAgencyId(entity.getOrgId()); |
|
|
|
try { |
|
|
|
natInfoScanTask(formDTO); |
|
|
|
}catch (Exception ee){ |
|
|
|
log.error(ee.getMessage()); |
|
|
|
throw new EpmetException(ee.getMessage()); |
|
|
|
}finally { |
|
|
|
updateSync(entity.getId(),OPERATION_STATUS_FINISH); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
}while (CollectionUtils.isNotEmpty(waitList)); |
|
|
|
} |
|
|
|
|
|
|
|
@Transactional(rollbackFor = Exception.class) |
|
|
|
public void updateSync(String id,String status){ |
|
|
|
LambdaUpdateWrapper<IcSyncJobEntity> qwUpdate = new LambdaUpdateWrapper<>(); |
|
|
|
qwUpdate.eq(BaseEpmetEntity::getId,id) |
|
|
|
.set(IcSyncJobEntity::getOperationStatus,status) |
|
|
|
.set(BaseEpmetEntity::getUpdatedTime, new Date()); |
|
|
|
icSyncJobDao.update(null,qwUpdate); |
|
|
|
} |
|
|
|
@Transactional(rollbackFor = Exception.class) |
|
|
|
public void insertSync(IcSyncJobEntity e){ |
|
|
|
icSyncJobService.insert(e); |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* @Description 配置信息查询 |
|
|
|
* @param customerId |
|
|
|
* @param dataCode |
|
|
|
* @Author zxc |
|
|
|
* @Date 2022/11/8 10:41 |
|
|
|
*/ |
|
|
|
public List<DataSyncConfigDTO> getConfigData(String customerId, String dataCode){ |
|
|
|
List<DataSyncConfigDTO> allConfigList = baseDao.list(customerId, "open",dataCode); |
|
|
|
return CollectionUtils.isEmpty(allConfigList) ? new ArrayList<>() : allConfigList; |
|
|
|
} |
|
|
|
|
|
|
|
private void dataSyncYanTaiParallel(List<DataSyncConfigDTO> configList, DataSyncTaskParam formDTO) { |
|
|
|
if (CollectionUtils.isEmpty(configList)) { |
|
|
|
log.warn("dataSyncYanTaiParallel configList is null"); |
|
|
|
|
wangxianzhang
3 years ago