java - 如何处理失败的 future

Question

我正在与我无法控制的遗留库集成。

它定义了以下接口(interface):

interface Factory {
    Future<Void> subscribe(Context c);
}

这个“订阅”方法被不同的线程频繁调用。我关心“Future.get()”结果的唯一情况是它失败时，这样我就可以获取并处理异常。这不必发生在调用线程中。另外，在我的情况下，阻塞调用“Future.get()”线程的成本可能相当昂贵，因为即使成功，也可能需要几秒钟才能完成。

所以我的任务是以某种方式“后处理”所有这些 future，过滤失败的。基本上，我可以看到两种可能的方法:

方法#1:

获取 Future 的实例后，向外部执行器提交一个单独的 Runnable 来执行所需的操作:

    executor.submit(
        new Runnable(){
            @Override
            public void run() {
                try {
                    future.get();
                } catch(Exception e){
                    // process the exception
                }
            }
        }
    );

这种方法的缺点是我们仍然会在很长一段时间内阻塞线程。正如我所说，此代码片段将相当频繁地执行。

方法#2:

在获取 Future 的实例时，将其放入某个集合并指定一个单独的线程，该线程将定期运行该集合的元素来进行处理:

    while(true){
        Iterator<Future<Void>> iterator = collection.iterator();
        while(iterator.hasNext()){
            Future<Void> future = iterator.next();
            if(future.isDone()){
                try {
                    future.get();
                } catch(Exception e){
                    // process the exception
                } finally {
                    iterator.remove();
                }
            }
        }

        TimeUnit.MILLISECONDS.sleep(1000); // sleep
    }

你觉得怎么样？有没有更好的方法来解决这个问题？

Answer 1

由于您没有参与 Future 的创建，因此最佳选项不可用，请使用自定义的 Future 来触发 Future 本身的处理.

因此，对于您的情况，我推荐一种可能看起来像是您的两个选项的混合的模式。将 Future 添加到(线程安全)队列，并将 Runnable 提交给执行器，该执行器在循环中处理所有项目。因此，您可以通过配置Executor来限制线程数量，即不要有与Future一样多的线程，但仍然可以有多个并且没有让这些后处理线程始终保持 Activity 状态。

为了避免在重新排队未完成的项目时出现无限循环，请使用本地集合将待处理的项目与重新排队的项目分开:

static BlockingQueue<Future<?>> PENDING = …;
static int MAX_ITEMS_PER_JOB = …;
…
/*scheduling code …*/new Runnable() {
  public void run()
  {
    ArrayList<Future<?>> myLocalItems=new ArrayList<>();
    PENDING.drainTo(myLocalItems, MAX_ITEMS_PER_JOB);
    for(Future<?> f:myLocalItems) {
      if(!f.isDone()) PENDING.offer(f); // re-queue
      try {
        f.get();
      } catch(ExecutionException ex) {
        handleException(ex.getCause());
      }
    }
  }
};

所以这个Runnable会检查并处理有限数量的Future然后返回，所以如果项目很多，适合多次提交以进行并行处理正在等待处理，但如果等待处理的数量较少，也没有什么坏处，因为如果没有什么可做的，这些工作就不会闲置。它们甚至适合使用 ScheduledExecutorService 的 scheduleWithFixedDelay。