java - ThreadPoolExecutor异常通知

Question

您能帮我验证这段代码吗？我正在尝试同时执行一些文件下载任务。稍后，如果所有下载都成功，那么我应该对它们进行后处理。如果任何任务失败，我应该向调用者抛出异常，因为我不想发布处理半成功的结果。

这是我的自定义 ThreadPoolExecutor，我尝试在其中检查池中任何线程的任何异常

public class CustomExecutorePool extends ThreadPoolExecutor {

private AtomicBoolean errorFlag = new AtomicBoolean(false);

public CustomExecutorePool(int corePoolSize, int maximumPoolSize,
        long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);

}


@Override
protected void afterExecute(Runnable r, Throwable childThreadException){
    if(null != childThreadException){
        logger.error("Exception occurred in one of the child threads: >>> " + childThreadException.getMessage() );
        errorFlag.set(true);
    }
}

public AtomicBoolean isErrorPresent(){
    return errorFlag;
}

如您所见，我尝试使用 AtmociBoolean 标志，并在任何子线程出现异常时将其设置为 true。

这是我的主类中的片段，我尝试使用此信息:

ThreadPoolExecutor docFetchingThreadPool = new CustomExecutorePool(...)
    try {
        // populate the thread pool queue for all the runnables to fetch documents
        for(Document myDoc : list){
            docFetchingThreadPool.execute(myDoc);
        }

        logger.info("All docs added to the work queue");
        docFetchingThreadPool.shutdown();
        logger.info("Waiting to finish work queue task for 3 mins");
        docFetchingThreadPool.awaitTermination(3, RETREIVAL_WAIT_UNIT);


        if(docFetchingThreadPool.isTerminated() && 
                !(docFetchingThreadPool.isErrorPresent().get()){

            //sucessfull post-process all docs now
            //......
        }else {
            throw new MyException("Exception occured in fetching");
        }
    } catch (InterruptedException e) {
        logger.error("Interrupt exception received ", e);
        Thread.currentThread().interrupt();
        throw new MyException("Interrupt exception");
    } finally {
        if(null != docFetchingThreadPool){
            logger.info("Force shoutdown for residual threads if any...");
            docFetchingThreadPool.shutdownNow();
        }
    }

您能否告诉我这看起来是否正常，或者错误 boolean 标志是否会出现任何同步问题，或者这是否是正确的方法。 我知道在执行程序服务上使用了 invokeall 函数，但仍在争论哪一个是处理任何异常场景的更简洁的方法。

感谢您的宝贵时间，

Answer 1

您不应该继承 ThreadPoolExecutor 的子类。特别是，这使得线程池只能用于一组任务的一次执行。

当您向线程池提交任务时，您会从线程池中获取 Future。因此，您只需迭代返回的 futures，get() 它们的结果，如果任何 get 方法抛出 ExecutionException(意味着任务本身抛出了异常，包装在其中)，则执行您想要的任何操作执行异常)。