取消由执行者服务控制的ComppletableFuture

Question

我有一个执行器服务，它将计算出的数据转发到可完成的未来：

class DataRetriever {
    private final ExecutorService service = ...;

    public CompletableFuture<Data> retrieve() {
        final CompletableFuture<Data> future = new CompletableFuture<>();

        service.execute(() -> {
            final Data data = ... fetch data ...
            future.complete(data);
        });
        return future;
    }
}

我希望客户端/用户能够取消任务：

final DataRetriever retriever = new DataRetriever();
final CompletableFuture<Data> future = retriever().retrieve();

future.cancel(true);

这不起作用，因为这将取消外部的可完成未来，而不是executor服务中计划的内部未来。

是否有可能将外部未来的取消（）传播到内部未来？

Answer 1

使用我的Tascalate并发库，您的代码可以重写如下：

class DataRetriever {
  private final ExecutorService service = ...;

  public Promise<Data> retrieve() {
    return CompletableTask.supplyAsync(() -> {
      final Data data = ... fetch data ...
      return data;
    }, service);
  }
}

Promise和CompletableTask，都是我的库中的类，您可以在我的博客中阅读更多内容

Answer 2

匿名用户

Pillar提到的另一个解决方案是扩展< code>CompletableFuture。这里有一个方法与您现有的代码非常相似。它还处理异常，这是一个不错的好处。

class CancelableFuture<T> extends CompletableFuture<T> {
    private Future<?> inner;

    /**
     * Creates a new CancelableFuture which will be completed by calling the
     * given {@link Callable} via the provided {@link ExecutorService}.
     */
    public CancelableFuture(Callable<T> task, ExecutorService executor) {
        this.inner = executor.submit(() -> complete(task));
    }

    /**
     * Completes this future by executing a {@link Callable}. If the call throws
     * an exception, the future will complete with that exception. Otherwise,
     * the future will complete with the value returned from the callable.
     */
    private void complete(Callable<T> callable) {
        try {
            T result = callable.call();
            complete(result);
        } catch (Exception e) {
            completeExceptionally(e);
        }
    }

    @Override
    public boolean cancel(boolean mayInterrupt) {
        return inner.cancel(mayInterrupt) && super.cancel(true);
    }
}

然后，在DataRetriever中，您只需执行以下操作：

public CompletableFuture<Data> retrieve() {
    return new CancelableFuture<>(() -> {... fetch data ...}, service);
}

Answer 3

CompletableFuture#cancel实际上仅用于将CompletableFuture标记为已取消。它不会通知正在执行的任务停止，因为它与任何此类任务都没有关系。

javadoc 暗示了这一点

mayInterruptIfRunning - 此值在此实现中没有影响，因为中断不用于控制处理。

在你的示例中，内在的未来是一个未来，而不是一个CompletableFuture，它确实与执行的Runnable（或Callable）有关系。在内部，由于它知道任务正在哪个线程上执行，因此它可以向它发送中断以尝试停止它。

一种选择是返回某种元组（例如一些POJO），该元组提供对您的CompletableFuture和由ExecutorService#提交返回的Future的引用。如果需要，您可以使用Future来取消。您必须记住取消或完成您的CompletableFuture，以便您的代码的其他部分不会永远保持阻塞/饥饿状态。