23

我有以下代码片段，它基本上扫描了需要执行的任务列表，然后将每个任务交给执行者执行。

反过来JobExecutor创建另一个执行程序（用于执行数据库操作...读取和写入数据到队列）并完成任务。

JobExecutorFuture<Boolean>为提交的任务返回一个。当其中一项任务失败时，我想优雅地中断所有线程并通过捕获所有异常来关闭执行程序。我需要做哪些改变？

public class DataMovingClass {
    private static final AtomicInteger uniqueId = new AtomicInteger(0);

  private static final ThreadLocal<Integer> uniqueNumber = new IDGenerator();   

  ThreadPoolExecutor threadPoolExecutor  = null ;

   private List<Source> sources = new ArrayList<Source>();

    private static class IDGenerator extends ThreadLocal<Integer> {
        @Override
        public Integer get() {
            return uniqueId.incrementAndGet();
        }
  }

  public void init(){

    // load sources list

  }

  public boolean execute() {

    boolean succcess = true ; 
    threadPoolExecutor = new ThreadPoolExecutor(10,10,
                10, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(1024),
                new ThreadFactory() {
                    public Thread newThread(Runnable r) {
                        Thread t = new Thread(r);
                        t.setName("DataMigration-" + uniqueNumber.get());
                        return t;
                    }// End method
                }, new ThreadPoolExecutor.CallerRunsPolicy());

     List<Future<Boolean>> result = new ArrayList<Future<Boolean>>();

     for (Source source : sources) {
                    result.add(threadPoolExecutor.submit(new JobExecutor(source)));
     }

     for (Future<Boolean> jobDone : result) {
                try {
                    if (!jobDone.get(100000, TimeUnit.SECONDS) && success) {
                        // in case of successful DbWriterClass, we don't need to change
                        // it.
                        success = false;
                    }
                } catch (Exception ex) {
                    // handle exceptions
                }
            }

  }

  public class JobExecutor implements Callable<Boolean>  {

        private ThreadPoolExecutor threadPoolExecutor ;
        Source jobSource ;
        public SourceJobExecutor(Source source) {
            this.jobSource = source;
            threadPoolExecutor = new ThreadPoolExecutor(10,10,10, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(1024),
                    new ThreadFactory() {
                        public Thread newThread(Runnable r) {
                            Thread t = new Thread(r);
                            t.setName("Job Executor-" + uniqueNumber.get());
                            return t;
                        }// End method
                    }, new ThreadPoolExecutor.CallerRunsPolicy());
        }

        public Boolean call() throws Exception {
            boolean status = true ; 
            System.out.println("Starting Job = " + jobSource.getName());
            try {

                        // do the specified task ; 


            }catch (InterruptedException intrEx) {
                logger.warn("InterruptedException", intrEx);
                status = false ;
            } catch(Exception e) {
                logger.fatal("Exception occurred while executing task "+jobSource.getName(),e);
                status = false ;
            }
           System.out.println("Ending Job = " + jobSource.getName());
            return status ;
        }
    }
}
4

3 回答 3

19

当您向执行者提交任务时，它会返回一个FutureTask实例。

FutureTask.get()将重新抛出任务抛出的任何异常作为ExecutorException.

因此，当您遍历List<Future>每个并调用 get 时，捕获ExecutorException并调用有序关闭。

于 2010-03-31T16:51:38.600 回答
14

由于您正在提交任务ThreadPoolExecutor，因此异常被FutureTask.

看看这段代码

**Inside FutureTask$Sync**

void innerRun() {
    if (!compareAndSetState(READY, RUNNING))
        return;

  runner = Thread.currentThread();
    if (getState() == RUNNING) { // recheck after setting thread
        V result;
       try {
            result = callable.call();
        } catch (Throwable ex) {
           setException(ex);
            return;
        }
       set(result);
    } else {
        releaseShared(0); // cancel
    }

}

protected void setException(Throwable t) {
   sync.innerSetException(t);
}

从上面的代码可以清楚地看出setException方法 catch Throwable。由于这个原因，FutureTask如果您使用“ submit()”方法，则会吞下所有异常ThreadPoolExecutor

根据java文档，您可以扩展afterExecute()方法ThreadPoolExecutor

protected void afterExecute(Runnable r,
                            Throwable t)

根据文档的示例代码：

class ExtendedExecutor extends ThreadPoolExecutor {
   // ...
   protected void afterExecute(Runnable r, Throwable t) {
     super.afterExecute(r, t);
     if (t == null && r instanceof Future<?>) {
       try {
         Object result = ((Future<?>) r).get();
       } catch (CancellationException ce) {
           t = ce;
       } catch (ExecutionException ee) {
           t = ee.getCause();
       } catch (InterruptedException ie) {
           Thread.currentThread().interrupt(); // ignore/reset
       }
     }
     if (t != null)
       System.out.println(t);
   }
 }

Exceptions你可以通过三种方式捕捉

  1. Future.get()正如接受的答案中所建议的那样
  2. 将整个run()或call()方法包装在try{}catch{}Exceptoion{}块中
  3. 如上所示afterExecute的方法的覆盖 ThreadPoolExecutor

要优雅地中断其他线程，请查看以下 SE 问题：

如何停止下一个线程在 ScheduledThreadPoolExecutor 中运行

如何强制关闭java ExecutorService

于 2016-01-06T18:45:16.417 回答
3

子类化ThreadPoolExecutor并覆盖它的protected afterExecute (Runnable r, Throwable t)方法。

如果您通过java.util.concurrent.Executors便利类（您不是）创建线程池，请查看其源代码以了解它是如何调用 ThreadPoolExecutor的。

于 2013-04-14T09:20:49.797 回答