Commit de2b0b9b by huangfusuper

分布式锁

parent 445c18e6
...@@ -42,8 +42,7 @@ public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunH ...@@ -42,8 +42,7 @@ public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunH
Long sleepTime = 0L; Long sleepTime = 0L;
try{ try{
//加锁 60秒后超时 //加锁 60秒后超时
boolean trlock = RedissLockUtil.trlock(lockName, TimeUnit.SECONDS, 1); if (RedissLockUtil.trlock(lockName, TimeUnit.SECONDS, 1)) {
if (trlock) {
log.debug("------------{},加锁成功,锁名称为{}----------",threadName,lockName); log.debug("------------{},加锁成功,锁名称为{}----------",threadName,lockName);
dateAligned(CYCLE_INTERVAL,threadName); dateAligned(CYCLE_INTERVAL,threadName);
//调用业务操作 //调用业务操作
......
...@@ -57,21 +57,15 @@ public class MakeUpFlowThreadRunHelper extends BaseThreadRunHelper { ...@@ -57,21 +57,15 @@ public class MakeUpFlowThreadRunHelper extends BaseThreadRunHelper {
*/ */
Map<Integer, WaitingRecord> nextRunFlow = allByTriggerTime.stream() Map<Integer, WaitingRecord> nextRunFlow = allByTriggerTime.stream()
.collect(Collectors.toMap(WaitingRecord::getFlowId, Function.identity(), BinaryOperator.minBy(Comparator.comparingInt(WaitingRecord::getWaitOrder)))); .collect(Collectors.toMap(WaitingRecord::getFlowId, Function.identity(), BinaryOperator.minBy(Comparator.comparingInt(WaitingRecord::getWaitOrder))));
Random random = new Random();
nextRunFlow.forEach((key,value) ->{ for (WaitingRecord value : nextRunFlow.values()) {
/*保证数据为串行*/
try {
Thread.sleep(random.nextInt(101)+100);
} catch (InterruptedException e) {
e.printStackTrace();
}
boolean runRecordingIsRunning = runRecordingService.findRunRecordingIsRunning(value.getFlowId()); boolean runRecordingIsRunning = runRecordingService.findRunRecordingIsRunning(value.getFlowId());
log.info("-------{}的运行状态为{}",value,runRecordingIsRunning); log.info("-------{}的运行状态为{}",value,runRecordingIsRunning);
log.info("==================================================的运行状态为{}======================================" ,runRecordingIsRunning); log.info("==================================================的运行状态为{}======================================" ,runRecordingIsRunning);
if(!runRecordingIsRunning){ if(!runRecordingIsRunning){
runRecordingAndJobTaskService.updateRunRecordingAndTask(value); runRecordingAndJobTaskService.updateRunRecordingAndTask(value);
} }
}); }
return UNIVERSAL_WAIT_TIME; return UNIVERSAL_WAIT_TIME;
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment