这个想法是任务正在推送到公共对象TaskCompleteEvent。如果他们推送错误,调度程序将停止,所有任务将停止。
您可以在“错误”和“成功”映射中检查每个任务迭代的结果。
public class SchedulerTest {
@Test
public void scheduler() throws InterruptedException {
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
TaskCompleteEvent taskCompleteEvent = new TaskCompleteEvent(scheduledExecutorService);
Runnable task1 = () -> {
int num = new Random().nextInt(100);
if (num < 5) {
taskCompleteEvent.message("task1-"+UUID.randomUUID().toString(),"Num "+num+" was obatined. Breaking all the executions.",true);
}
};
Runnable task2 = () -> {
int num = new Random().nextInt(100);
taskCompleteEvent.message("task2-"+UUID.randomUUID().toString(),num < 50,false);
};
scheduledExecutorService.scheduleAtFixedRate(task1,1,TimeUnit.SECONDS);
scheduledExecutorService.scheduleAtFixedRate(task2,TimeUnit.SECONDS);
scheduledExecutorService.awaitTermination(60,TimeUnit.SECONDS);
System.out.println("Success: "+taskCompleteEvent.getSuccess());
System.out.println("Errors: "+taskCompleteEvent.getErrors());
System.out.println("Went well?: "+taskCompleteEvent.getErrors().isEmpty());
}
public static class TaskCompleteEvent {
private final ScheduledExecutorService scheduledExecutorService;
private final Map<String,Object> errors = new LinkedHashMap<>();
private final Map<String,Object> success = new LinkedHashMap<>();
public TaskCompleteEvent(ScheduledExecutorService scheduledExecutorService) {
this.scheduledExecutorService = scheduledExecutorService;
}
public synchronized void message(String id,Object response,boolean error) {
if (error) {
errors.put(id,response);
scheduledExecutorService.shutdown();
} else {
success.put(id,response);
}
}
public synchronized Map<String,Object> getErrors() {
return errors;
}
public synchronized Map<String,Object> getSuccess() {
return success;
}
}
}
,
您只需要添加一个其他任务,该任务的任务是监视所有其他正在运行的任务-并且当任何被监视的任务失败时,它们需要设置一个刺客可以检查的信号灯(标志)。 / p>
ScheduledExecutorService executor = (ScheduledExecutorService) Executors.newScheduledThreadPool(2);
// INSTANTIATE THE REMOTE-FILE-MONITOR:
RemoteFileMonitor monitor = new RemoteFileMonitor(remotesource,localtarget);
// THIS TimerTask PERIODICALLY TRIGGERS THE RemoteFileMonitor:
TimerTask remote = new TimerTask() {
// RUN FORREST... RUN !
public void run() {
try {
kae.trace("TimerTask::run() --> Calling RemoteFileMonitor.check()");
monitor.check();
} catch (Exception ex) {
// NULL TRAP: ALLOWS US TO CONTINUE AND RETRY:
}
}
};
// THIS TimerTask PERIODICALLY TRIES TO KILL THE REMOTE-FILE-MONITOR:
TimerTask assassin = new TimerTask() {
// WHERE DO BAD FOLKS GO WHEN THEY DIE ?
private final LocalDateTime death = LocalDateTime.now().plus(ConfigurationOptions.getPollingCycleTime(),ChronoUnit.MINUTES);
// RUN FORREST... RUN !
public void run() {
// IS THERE LIFE AFTER DEATH ???
if (LocalDateTime.now().isAfter(death)) {
// THEY GO TO A LAKE OF FIRE AND FRY:
kae.error(ReturnCode.MONITOR_POLLING_CYCLE_EXCEEDED);
}
}
};
// SCHEDULE THE PERIODIC EXECUTION OF THE RemoteFileMonitor: (remote --> run() monitor --> check())
executor.scheduleAtFixedRate(remote,delay,interval,TimeUnit.MINUTES);
// SCHEDULE PERIODIC ASSASSINATION ATTEMPTS AGAINST THE RemoteFileMonitor: (assassin --> run() --> after death --> die())
executor.scheduleAtFixedRate(assassin,60L,TimeUnit.SECONDS);
// LOOP UNTIL THE MONITOR COMPLETES:
do {
try {
// I THINK I NEED A NAP:
Thread.sleep(interval * 10);
} catch (InterruptedException e) {
// FAIL && THEN cleanexit();
kae.error(ReturnCode.MONITORING_ERROR,"Monitoring of the XXXXXX-Ingestion site was interrupted");
}
// NOTE: THE MONITOR IS SET TO 'FINISHED' WHEN THE DONE-File IS DELIVERED AND RETRIEVED:
} while (monitor.isNotFinished());
// SHUTDOWN THE MONITOR TASK:
executor.shutdown();
本文链接:https://www.f2er.com/3128198.html