wangxianzhang
3 years ago
7 changed files with 253 additions and 45 deletions
@ -0,0 +1,126 @@ |
|||
package com.epmet.processor; |
|||
|
|||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; |
|||
import com.epmet.commons.tools.distributedlock.DistributedLock; |
|||
import com.epmet.commons.tools.exception.ExceptionUtils; |
|||
import com.epmet.commons.tools.redis.RedisUtils; |
|||
import com.epmet.constant.EpidemicConstant; |
|||
import com.epmet.dao.IcSyncJobDao; |
|||
import com.epmet.entity.IcSyncJobEntity; |
|||
import com.epmet.service.DataSyncConfigService; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.redisson.api.RLock; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.scheduling.annotation.Scheduled; |
|||
import org.springframework.stereotype.Component; |
|||
import org.springframework.util.CollectionUtils; |
|||
import org.springframework.util.StringUtils; |
|||
|
|||
import java.util.List; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
import static com.epmet.constant.EpidemicConstant.JOB_TYPE_NAT; |
|||
|
|||
// 烟台核酸检测数据同步处理器
|
|||
@Component |
|||
@Slf4j |
|||
public class YanTaiNatSyncProcessor { |
|||
|
|||
public static final int MAX_EXECUTING_COUNT = 3; |
|||
|
|||
@Autowired |
|||
private ExecutorService executorService; |
|||
|
|||
@Autowired |
|||
private IcSyncJobDao icSyncJobDao; |
|||
|
|||
@Autowired |
|||
private DataSyncConfigService dataSyncConfigService; |
|||
|
|||
@Autowired |
|||
private DistributedLock distributedLock; |
|||
|
|||
@Autowired |
|||
RedisUtils redisUtils; |
|||
|
|||
/** |
|||
* 定时扫描和执行同步任务 |
|||
* 10s扫一次库 |
|||
* |
|||
* @author wxz |
|||
* @date 2022/11/8 下午5:42 |
|||
*/ |
|||
@Scheduled(cron = "0/10 * * * * ? ") |
|||
public void scanJobs() { |
|||
log.info("【异步数据更新】开始同步任务"); |
|||
|
|||
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); |
|||
List<IcSyncJobEntity> executingJobList = icSyncJobDao.selectList(executingListQuery); |
|||
|
|||
if (!CollectionUtils.isEmpty(executingJobList) && executingJobList.size() >= MAX_EXECUTING_COUNT) { |
|||
// 最多只允许同时3条线程运行
|
|||
return; |
|||
} |
|||
|
|||
int executingCount = executingJobList.size(); |
|||
// 还可以运行几条线程
|
|||
int leftCount = MAX_EXECUTING_COUNT - executingCount; |
|||
|
|||
RLock lock = null; |
|||
try { |
|||
lock = distributedLock.getLock("data:sync:" + JOB_TYPE_NAT, 60L, 60L, TimeUnit.SECONDS); |
|||
// 查询可执行的任务列表，并且异步执行
|
|||
List<IcSyncJobEntity> icSyncJobToExec = icSyncJobDao.selectExecutableJobList( |
|||
EpidemicConstant.JOB_TYPE_NAT, |
|||
EpidemicConstant.OPERATION_STATUS_WAITING, |
|||
leftCount); |
|||
|
|||
if (!CollectionUtils.isEmpty(icSyncJobToExec)) { |
|||
// 异步提交任务
|
|||
for (IcSyncJobEntity jobEntity : icSyncJobToExec) { |
|||
|
|||
updateJobStatus(jobEntity.getId(), EpidemicConstant.OPERATION_STATUS_PROCESSING); |
|||
executorService.submit(() -> { |
|||
// 将此任务状态修改为执行中
|
|||
|
|||
dataSyncConfigService.execSyncByJobProcessor(jobEntity); |
|||
|
|||
// 更新任务状态为结束
|
|||
updateJobStatus(jobEntity.getId(), EpidemicConstant.OPERATION_STATUS_FINISH); |
|||
}); |
|||
} |
|||
} |
|||
} catch (Exception e) { |
|||
log.error("【异步数据更新】出错:{}", ExceptionUtils.getErrorStackTrace(e)); |
|||
} finally { |
|||
if (lock != null) { |
|||
lock.unlock(); |
|||
} |
|||
} |
|||
} |
|||
|
|||
/** |
|||
* 更新任务状态 |
|||
* @author wxz |
|||
* @date 2022/11/8 下午8:25 |
|||
* @param id |
|||
* @param status |
|||
|
|||
*/ |
|||
private void updateJobStatus(String id, String status) { |
|||
LambdaQueryWrapper<IcSyncJobEntity> query = new LambdaQueryWrapper<>(); |
|||
query.eq(IcSyncJobEntity::getId, id); |
|||
|
|||
IcSyncJobEntity updateEntity = new IcSyncJobEntity(); |
|||
updateEntity.setOperationStatus(status); |
|||
icSyncJobDao.update(updateEntity, query); |
|||
} |
|||
|
|||
} |
|||
Loading…
Reference in new issue