java - 在设计异步 Java API 时，如何确保整个 CompletableFuture 链由内部线程池执行？

Question

我正在编写一个库，该库提供多个返回 CompletableFutures 的异步方法。该库有一个内部线程池，用于执行异步方法的计算工作。

我想确保满足以下两个要求:

1. 返回的 CompletableFuture 不是由内部线程完成的，因此库的内部线程池永远不会执行我库外部的 CompletableFuture 链
2. 我的异步方法的所有计算都由内部线程池执行，而不是由用户线程(即方法调用者的线程)执行

假设库有以下阻塞方法

Data request(Address address) {
  Message request = encode(address);
  Message response = sendAndReceive(request);
  Data responseData = decode(response);
  return responseData;
}

和相应的异步方法

  CompletableFuture<Data> requestAsync(Address address) {
    return CompletableFuture.supplyAsync(() -> encode(address), internalThreadPool)
        .thenCompose(request -> sendAndReceiveAsync(request))
        .thenApply(response -> decode(response));
  }

第一个要求是通过添加链接 .whenCompleteAsync((v,t) -> {}) 来满足的，如 this answer 中所述。 .

但是要满足第二个要求需要做什么呢？

Answer 1

Sergey Kuksenko 已经讨论了第二个要求的解决方案 here并已在 Java 11 的 HttpClient 实现中实现。

不满足要求，因为不能保证decode(response)是由内部线程执行的。如果编码和 sendAndReceiveAsync 快速完成，decode 实际上可以由调用者的线程执行。

这个问题可以通过引入 CompletableFuture startCf 来启动 CF 链来解决。

因此完整的解决方案可能如下所示

CompletableFuture<Data> requestAsyncFixedAll(Address address) {
  CompletableFuture<Void> startCf = new CompletableFuture<>();
  CompletableFuture<Data> dataCf =  startCf.thenApplyAsync(v -> encode(address), internalThreadPool)
    .thenCompose(request -> sendAndReceiveAsync(request))
    .thenApply(response -> decode(response)).whenCompleteAsync((v, t) -> {});
  startCf.complete(null);
  return dataCf;
}