如何使ScheduledThreadPool报告错误？

Question

在经历了痛苦的调试经历之后,我追踪了这个问题:ScheduledThreadPool不报告任务是否失败,并且不执行再次失败的任务.因此,很难跟踪周期性作业的活跃性,而不是用其他周期性任务(通过死人的开关或者ScheduledFuture)来检查它们.

现在我们可以提供ScheduledThreadPool一个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

然后它挂起(按设计).

我总是可以尝试抓住整个任务,但这感觉很难看; 在UncaughtExceptionHandler应该这样做了!

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

Answer 1

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

一个简单的方法就是包装你的runnable.

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!
   ...