java - 如何让ScheduledThreadPool报错？

Question

在经历了痛苦的调试之后，我找到了这个问题:ScheduledThreadPool 不报告任务失败，并且不执行再次失败的任务。因此，很难跟踪周期性作业的活跃度，除非用其他周期性任务检查它们(通过 dead man 开关或 ScheduledFuture)。

现在我们可以将 ScheduledThreadPool 和 UncaughtExceptionHandler 传递给一个 UncaughtExceptionHandler，但即使那样似乎也行不通:

import java.util.concurrent.*;

class Test {
  public static void main(String[] args) {
    final ThreadFactory tf = new ThreadFactory() {
      private final ThreadFactory delegate = Executors.defaultThreadFactory();

      @Override public Thread newThread(final Runnable r) {
        final Thread res = delegate.newThread(r);
        res.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
          @Override
          public void uncaughtException(final Thread t, final Throwable e) {
            e.printStackTrace();
          }
        });
        return res;
      }
    };
    final ScheduledThreadPoolExecutor exec = new ScheduledThreadPoolExecutor(1, tf);

    final Runnable task = new Runnable() {
      private int c = 0;

      @Override
      public void run() {
        if ( c++ == 5 ) {
          throw new ArrayIndexOutOfBoundsException("Runtime error!");
        }

        System.out.println("Reached " + c);
      }
    };

    exec.scheduleWithFixedDelay(task, 1, 1, TimeUnit.SECONDS);
  }
}

这个程序的输出很简单(Oracle Java SE(64 位服务器)1.7.0_06-b24)

Reached 1
Reached 2
Reached 3
Reached 4
Reached 5

然后它挂起(按设计)。

我总是可以 try-catch 整个任务，但这感觉很难看； UncaughtExceptionHandler 应该已经做到了!

是否有针对此问题的 API 解决方案？我做错了什么，还是一个错误？

Answer 1

货币线程池捕获所有异常，然后放入 Future 对象中供您检查。 UncaughtExceptionHandler 仅用于线程未捕获并杀死线程的异常，在这种情况下，它仅用于线程池代码抛出的异常。

解决此问题的一个简单方法是包装您的可运行对象。

public class ExceptionHandlingScheduledExecutor extends ScheduledThreadPoolExecutor {
    private final Thread.UncaughtExceptionHandler ueh;

    public ExceptionHandlingScheduledExecutor(int corePoolSize, Thread.UncaughtExceptionHandler ueh) {
        super(corePoolSize);
        this.ueh = ueh;
    }

    @Override
    public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
        return super.schedule(wrap(command), delay, unit);
    }

    @Override
    public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
        return super.schedule(wrap(callable), delay, unit); 
    }

    @Override
    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
        return super.scheduleAtFixedRate(wrap(command), initialDelay, period, unit);
    }

    @Override
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
        return super.scheduleWithFixedDelay(wrap(command), initialDelay, delay, unit);
    }

    @Override
    public void execute(Runnable command) {
        super.execute(wrap(command));
    }

    @Override
    public Future<?> submit(Runnable task) {
        return super.submit(wrap(task));
    }

    @Override
    public <T> Future<T> submit(Runnable task, T result) {
        return super.submit(wrap(task), result);
    }

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        return super.submit(wrap(task));
    }

    private Runnable wrap(final Runnable runnable) {
        return new Runnable() {
            @Override
            public void run() {
                try {
                    runnable.run();
                } catch (final Throwable t) {
                    ueh.uncaughtException(Thread.currentThread(), t);
                    throw t;
                }
            }
        };
    }

    private <T> Callable<T> wrap(final Callable<T> callable) {
        return new Callable<T>() {
            @Override
            public T call() throws Exception {
                try {
                    return callable.call();
                } catch (Throwable t) {
                    ueh.uncaughtException(Thread.currentThread(), t);
                    throw t;
                }
            }
        };
    }
}

您可以子类化 ThreadPoolExecutor 以透明地执行此操作。

---

您也可以使用缓存的线程池来处理异常，但这样比较复杂。

以透明方式使用返回的 Future 的一种方法是子类化 ScheduledThreadPoolExecutor(或任何 Executor，就此而言):

class MyScheduledExecutor extends ScheduledThreadPoolExecutor {
  private final Thread.UncaughtExceptionHandler ueh;
  private final ExecutorService futureService = Executors.newCachedThreadPool();

  public MyScheduledExecutor(int corePoolSize, Thread.UncaughtExceptionHandler ueh) {
    super(corePoolSize);
    this.ueh = ueh;
  }

  // Copy other constructors

  @Override
  public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                   long initialDelay,
                                                   long delay,
                                                   TimeUnit unit) {
    final ScheduledFuture<?> f = super.scheduleWithFixedDelay(command, initialDelay, delay, unit);
    futureService.submit(new Runnable() {
      @Override
      public void run() {
        try {
          f.get();
        } catch (Throwable t ) {
          ueh.uncaughtException(null, t.getCause());
        }
      }
    };

    return f;
  }

  // Do similarly for other submit/schedule methods
}

然后像这样使用它:

final ScheduledThreadPoolExecutor exec = new MyScheduledExecutor(1, new Thread.UncaughtExceptionHandler() {
      @Override
      public void uncaughtException(final Thread t, final Throwable e) {
        e.printStackTrace();
      }
    });

现在输出如我们所愿:

Reached 1
Reached 2
Reached 3
Reached 4
Reached 5
java.lang.ArrayIndexOutOfBoundsException: Runtime error!
   ...