番石榴期货等待回调

Question

我有一个未来的列表，在每个未来完成后，我有一个应该执行的回调。

我在使用期货。successfulAsList检查是否所有期货都已完成。然而，这并没有考虑回调的完成。

有没有办法确保回调完成？

我可以用期货代替回调。转换为包装到另一个未来，并检查其是否完成。然而，这样一来，我就无法访问包装好的将来引发的运行时异常。

ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(20));

List<ListenableFuture<Object>> futures = new ArrayList<>();

for (int i = 1; i <= 20; i++) {
  final int x = i * 100;

  ListenableFuture<Object> future = service.submit(new Callable() {
    @Override
    public Object call() throws Exception {
      Thread.sleep(10000 / x);

      return x;
    }
  });

  futures.add(future);

  Futures.addCallback(future, new FutureCallback<Object>() {

    @Override
    public void onFailure(Throwable t) {
      t.printStackTrace();
    }

    @Override
    public void onSuccess(Object x) {
      try {Thread.sleep((Integer) x * 10);} catch (Exception e) {}

      System.out.println(x);
    }
  });
}

ListenableFuture<List<Object>> listFuture = Futures
    .successfulAsList(futures);
System.out.println("Waiting...");
System.out.println(listFuture.get());
System.out.println("Done");

Answer 1

谢谢，这工作！

ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(20));

List<ListenableFuture<Void>> futures = new ArrayList<>();

for (int i = 1; i <= 20; i ++) {
  final int x = i * 100;

  ListenableFuture<Object> future = service.submit(new Callable(){
    @Override
    public Object call() throws Exception {
      Thread.sleep(10000 / x);

      return x;
    }
  });

  //Blank runnable to evaluate write completion
  Runnable callback = new Runnable(){
    @Override
    public void run() {
      //do nothing
    }
  };

  final ListenableFutureTask<Void> callbackFuture = ListenableFutureTask.create(callback, null);

  futures.add(callbackFuture);

  Futures.addCallback(future, new FutureCallback<Object>() {

    @Override
    public void onFailure(Throwable t) {
      try {
        t.printStackTrace();
      }
      finally {
        callbackFuture.run();
      }
    }

    @Override
    public void onSuccess(Object x) {
      try {
        try {Thread.sleep((Integer)x*10);}catch(Exception e){}

        System.out.println(x);
      }
      finally {
        callbackFuture.run();
      }
    }
  });
}

ListenableFuture<List<Void>> listFuture = Futures.successfulAsList(futures);
System.out.println("Waiting...");
System.out.println(listFuture.get());
System.out.println("Done");

Answer 2

如果您为每个回调创建另一个未来并确保它将在回调中完成，怎么样？

// create "callback" future here
futures.add(callbackFuture);

Futures.addCallback(future, new FutureCallback<Object>() {

  @Override
  public void onFailure(Throwable t) {
    t.printStackTrace();
    // do something with callbackFuture
  }

  @Override
  public void onSuccess(Object x) {
    try {Thread.sleep((Integer) x * 10);} catch (Exception e) {}

    System.out.println(x);
    // do something with callbackFuture
  }
});

Answer 3

如果您只想阻止，直到您提交的N个任务的回调全部完成，您可以创建一个计数为N的倒计时锁存器（CountDownLatch）。然后在每次回调完成时（无论成功与否）调用倒计时锁存器（CountDownLatch），然后在要阻止的点调用它。

或者，您可以像在回答中一样做一些事情，但不是使用ListenableFutureTask