处理Java ExecutorService任务中的异常

Question

我正在尝试使用Java的ThreadPoolExecutor类来运行具有固定数量线程的大量重量级任务.每个任务都有许多地方,在这些地方可能因异常而失败.

我已经进行了子类化,ThreadPoolExecutor并且我已经覆盖了该afterExecute方法,该方法应该在运行任务时提供任何未捕获的异常.但是,我似乎无法使其发挥作用.

例如:

public class ThreadPoolErrors extends ThreadPoolExecutor {
    public ThreadPoolErrors() {
        super(  1, // core threads
                1, // max threads
                1, // timeout
                TimeUnit.MINUTES, // timeout units
                new LinkedBlockingQueue<Runnable>() // work queue
        );
    }

    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        if(t != null) {
            System.out.println("Got an error: " + t);
        } else {
            System.out.println("Everything's fine--situation normal!");
        }
    }

    public static void main( String [] args) {
        ThreadPoolErrors threadPool = new ThreadPoolErrors();
        threadPool.submit( 
                new Runnable() {
                    public void run() {
                        throw new RuntimeException("Ouch! Got an error.");
                    }
                }
        );
        threadPool.shutdown();
    }
}

这个程序的输出是"一切都很好 - 情况正常!" 即使提交给线程池的唯一Runnable也会引发异常.有什么线索在这里发生了什么？

谢谢!

Answer 1

警告:应该注意,此解决方案将阻止调用线程.

如果要处理任务抛出的异常,那么通常最好使用Callable而不是Runnable.

Callable.call() 允许抛出已检查的异常,并将这些异常传播回调用线程:

Callable task = ...
Future future = executor.submit(task);
try {
   future.get();
} catch (ExecutionException ex) {
   ex.getCause().printStackTrace();
}

如果Callable.call()抛出一个异常,它将被包装在一个ExecutionException并被抛出Future.get().

这可能比继承更好ThreadPoolExecutor.如果异常是可恢复的,它还为您提供重新提交任务的机会.

Answer 2

来自文档:

注意:当任务明确地或通过诸如submit之类的方法包含在任务(例如FutureTask)中时,这些任务对象会捕获并维护计算异常,因此它们不会导致突然终止,并且内部异常不会传递给此方法.

当你提交一个Runnable时,它将被包含在Future中.

你的afterExecute应该是这样的:

public final class ExtendedExecutor extends ThreadPoolExecutor {

    // ...

    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        if (t == null && r instanceof Future<?>) {
            try {
                Future<?> future = (Future<?>) r;
                if (future.isDone()) {
                    future.get();
                }
            } catch (CancellationException ce) {
                t = ce;
            } catch (ExecutionException ee) {
                t = ee.getCause();
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
            }
        }
        if (t != null) {
            System.out.println(t);
        }
    }
}

Answer 3

这种行为的解释正好在afterExecute的javadoc中:

注意:当任务明确地或通过诸如submit之类的方法包含在任务(例如FutureTask)中时,这些任务对象会捕获并维护计算异常,因此它们不会导致突然终止,并且内部异常不会传递给此方法.

Answer 4

我通过将提交的runnable包装提交给执行程序来解决它.

CompletableFuture.runAsync(

        () -> {
                try {
                        runnable.run();
                } catch (Throwable e) {
                        Log.info(Concurrency.class, "runAsync", e);
                }
        },

        executorService
);

Answer 5

我正在使用VerboseRunnable来自jcabi-log的类,它会吞下所有异常并记录它们.非常方便,例如:

import com.jcabi.log.VerboseRunnable;
scheduler.scheduleWithFixedDelay(
  new VerboseRunnable(
    Runnable() {
      public void run() { 
        // the code, which may throw
      }
    },
    true // it means that all exceptions will be swallowed and logged
  ),
  1, 1, TimeUnit.MILLISECONDS
);

Answer 6

另一个解决方案是使用ManagedTask和ManagedTaskListener。

您需要一个实现ManagedTask接口的Callable或Runnable。

该方法getManagedTaskListener返回您想要的实例。

public ManagedTaskListener getManagedTaskListener() {

然后您在ManagedTaskListener中实现该taskDone方法：

@Override
public void taskDone(Future<?> future, ManagedExecutorService executor, Object task, Throwable exception) {
    if (exception != null) {
        LOGGER.log(Level.SEVERE, exception.getMessage());
    }
}

有关托管任务生命周期和侦听器的更多详细信息。