Java停止执行器服务，一旦他分配的任务因任何原因失败

Question

我需要某种服务，它可以同时运行一些任务，间隔1秒，持续1分钟。

如果其中一个任务失败，我想停止服务和与它一起运行的每一个任务，并显示出某种问题，否则，如果一分钟后一切顺利，服务将停止，并显示一切顺利。

例如，我有两个功能：

Runnable task1 = ()->{
int num = Math.rand(1,100);
if (num < 5){
throw new Exception("something went wrong with this task,terminate");
}
}
Runnable task2 = ()->{
int num = Math.rand(1,100)
return num < 50;
}

ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
task1schedule = scheduledExecutorService.scheduleAtFixedRate(task1, 1, 60, TimeUnit.SECONDS);
task2schedule = scheduledExecutorService.scheduleAtFixedRate(task2, 1, 60, TimeUnit.SECONDS);
if (!task1schedule || !task2schedule) scheduledExecutorService.shutdown();

关于我应该如何解决这个问题并使其尽可能通用，有什么想法吗？

Answer 1

这个想法是任务正在推送到一个公共对象TaskCompleteEvent。如果他们推送一个错误，调度程序将停止，所有任务都将停止。

您可以在映射"errors"one_answers"success"中检查每个任务迭代的结果。

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, 0, 1, TimeUnit.SECONDS);
scheduledExecutorService.scheduleAtFixedRate(task2, 0, 1, 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;
}
}
}

Answer 2

您只需要添加一个额外的任务，该任务的任务是监视所有其他正在运行的任务——当任何被监视的任务失败时，它们需要设置一个信号量(标志(，让刺客可以检查。

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, delay, 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();