0

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

这是我的自定义 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();
        }
    }

如果这看起来不错，请告诉我，或者错误布尔标志是否会有任何同步问题，或者这甚至是正确的方法。我知道在 executor 服务上使用 invokeall 函数，但仍在争论哪一个将是处理任何异常情况的更清洁的方法。

谢谢你的时间，

4

1 回答 1

1

你不应该继承 ThreadPoolExecutor。特别是，这使得线程池仅可用于一组任务的一次执行。

当您将任务提交到线程池时，您会Future从线程池中获得回馈。因此，您只需遍历返回的期货get()及其结果，并在任何 get 方法抛出 ExecutionException 时做任何您想做的事情（这意味着任务本身抛出了异常，包装在 ExecutionException 中）。

于 2013-07-05T06:09:33.437 回答