java - グアバ先物はコールバックを待つ

Question

先物のリストがあり、各先物が完了すると、実行されるコールバックがあります。

Futures.successfulAsList を使用して、すべての先物が完了したかどうかを確認しています。ただし、これはコールバックの完了を考慮していません。

コールバックが完了したことを確認する方法はありますか?

Callback の代わりに、Futures.transform を使用して別の Future にラップし、その完了を確認できます。ただし、これでは、ラップされた未来でスローされたランタイム例外にアクセスできません。

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

コールバックごとに別の未来を作成し、それがコールバック内で完了することを確認する場合はどうでしょうか。

// 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 2

睡眠なしの実現：

    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.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);
            }
        });

        /* all Callbacks added in one list (ExecutionList) and executed by order. If not defined 3d argument (Executor)
           then callbacks executed sequentially at task thread.
         */
        final SettableFuture<Object> lastCalledFuture = SettableFuture.create();
        Futures.addCallback(future, new FutureCallback<Object>() {
            @Override
            public void onSuccess(Object result) {
                lastCalledFuture.set(result);
            }

            @Override
            public void onFailure(Throwable t) {
                lastCalledFuture.setException(t);
            }
        });
        futures.add(lastCalledFuture);
    }

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

Answer 3

ありがとう、これはうまくいきます！

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");