5

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

我正在使用 Futures.successfulAsList 检查所有期货是否已完成。但是，这并没有考虑到回调的完成。

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

而不是回调，我可以使用 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");
4

4 回答 4

5

如果您只想阻塞直到您提交的 N 个任务的回调全部完成，您可以CountDownLatch使用 N 的 a创建一个count。然后countDown()在每个回调完成时调用它（无论是成功还是失败），await()然后在您想要阻止。

或者，您可以像在答案中那样做一些事情，但不要使用 aListenableFutureTask<Void>和 no-op Runnable，而是使用 aSettableFuture<Void>并在完成时调用set(null)它。

于 2013-07-16T17:54:08.410 回答
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
  }
});
于 2013-07-16T09:02:33.903 回答
0

谢谢，这行得通！

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");
于 2013-07-16T11:12:56.560 回答
0

不睡觉的实现：

    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");
于 2015-10-27T09:45:04.850 回答