来自 ScheduledExecutorService 的多个期货的异常处理

Question

我正在使用 ScheduledExecutorService，我将提交 Runnables 和计划的 Runnables 添加到其中（使用 scheduleWithFixedDelay）。 目的是让进程运行很长时间，因此我的所有可运行程序都没有定义的生命周期。我基本上希望主线程只对异常和中断做出反应。 计划任务很关键，例如产生热信号，因此如果任何线程抛出运行时异常，我想记录异常，中止所有其他线程并关闭程序。

我应该如何处理异常？除非我运行 Future.get()，否则 ScheduledExecutorService 会吞下所有异常。

如下所示，循环遍历期货是行不通的。如果期货列表中的第一项没有返回任何错误，这将阻止线程侦听可能返回错误的其他线程。

for (Future<?> future : futures) {
    future.get();
}

一个选项是循环通过期货询问它们是否完成，如下所示，但我不太喜欢这种解决方案。我需要添加一个线程睡眠，因此对异常的响应会延迟。

boolean allActive = true;
while (allActive) {
    for (Future<?> future : futures) {
        if (!future.isDone()) {
             allActive = false;
             break;
        }
    }
    Thread.sleep(50);
}

我还有什么其他选择？还是我处理问题的方法有误？ 我不应该使用 ScheduledExecutorService 并在我自己的线程中自己实现计划吗？

示例代码，尝试将顺序更改为未来列表！如果您在句柄 2 之前添加句柄，我想要您得到的行为，但列表的顺序无关紧要：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class Test {

    private static int i = 0;

    public static void main(String[] args) throws Exception {

        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

        Future<?> handle = scheduler.scheduleWithFixedDelay(new Runnable() {
            public void run() {
                System.out.println("No exception!");
                if (i > 2) {
                    System.out.println("Exception!");
                    throw new RuntimeException("foo");
                }
                i++;
            }
        }, 0, 500, TimeUnit.MILLISECONDS);

        Future<?> handle2 = scheduler.scheduleWithFixedDelay(new Runnable() {
            public void run() {
                System.out.println("Running!");
            }
        }, 0, 500, TimeUnit.MILLISECONDS);


        List<Future<?>> futures = new ArrayList<>();
        futures.add(handle2);
        futures.add(handle);

        try {
            for (Future<?> future : futures) {
                future.get();
            }
        } catch (Exception e) {
            scheduler.shutdownNow();
            System.out.println(scheduler.awaitTermination(1, TimeUnit.SECONDS));
            System.out.println("Shuwdown complete");
            e.printStackTrace();
        }
    }
}

Answer 1

您可以使用 Listener 或 Observer-Like 模式来做到这一点：

interface IFutureListener{
    void onException( Throwable t );
}


final IFutureListener errHandler = new IFutureListener(){
     @override public void onException( Throwable t ){
          // shutdown Service here
     }
};
// ...
Future<?> handle = scheduler.scheduleWithFixedDelay(new Runnable() {
        final IFutureListener callback = errHandler;
        public void run() {
            try{
            System.out.println("No exception!");
            if (i > 2) {
                System.out.println("Exception!");
                throw new RuntimeException("foo");
            }
            i++;
            }
            catch( Exception ex ){
                callback.onException(ex);
            }
        }
    }, 0, 500, TimeUnit.MILLISECONDS);

您可能仍然需要对此进行一些调整，但这就是它的要点。

@dimo414 在评论中写的 Guava 的 ListenableFuture 会给你类似的东西。但是，如果您不想/不允许使用 3rd 方，这是您可以自己实现的一种方式。

感谢@efekctive：我还建议记录异常。除非您确切地知道自己在做什么，否则它们几乎不应该被默默吞下。