java - 使用 "Real"Executor 的 Guava Futures.transform 与使用 DirectExecutor 的 transformAsync

Question

假设我有两个重量级的 IO 阻塞操作，findMatchInSomeDB() 和 getDetailsFromOtherDB(String objectKey)。此外，我希望它们在后台运行，并使用 Guava Futures 将它们链接在一起，因为一个取决于另一个 的结果(我知道这可以在一个简单的 Callable 中按顺序完成，但保持简单说明目的):

下面的 consumeChain 和 consumeChainAsync 方法之间是否有任何实际或细微差别？

import com.google.common.base.Function;
import com.google.common.util.concurrent.AsyncFunction;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;

public class Consumer
{
   private final Retriever retriever;
   private final ListeningExecutorService executorSvc;

   public Consumer(Retriever retriever, ListeningExecutorService executorSvc)
   {
      this.retriever = retriever;
      this.executorSvc = executorSvc;
   }

   private void consumeChain(String searchCriteria) throws Exception
   {
      ListenableFuture<String> futureMatch = executorSvc.submit(
         () -> retriever.findMatchInSomeDB(searchCriteria));

      Function<String, DataObject> keyToDataObj = objectKey ->
         retriever.getDetailsFromOtherDB(objectKey);

      // using "real" executor service so transform function runs
      // in the background
      ListenableFuture<DataObject> futureDataObj = Futures.transform(
         futureMatch, keyToDataObj, executorSvc);

      // ...do some other stuff in this thread...

      // do something with futureDataObj result
      futureDataObj.get();
   }

   private void consumeChainAsync(String searchCriteria) throws Exception
   {
      ListenableFuture<String> futureMatch = executorSvc.submit(
         () -> retriever.findMatchInSomeDB(searchCriteria));

      AsyncFunction<String, DataObject> keyToDataObj = objectKey ->
      {
         return executorSvc.submit(
            () -> retriever.getDetailsFromOtherDB(objectKey));
      };

      // using DirectExecutor because the transform function
      // only creates and submits a Callable
      ListenableFuture<DataObject> futureDataObj = Futures.transformAsync(
         futureMatch, keyToDataObj, MoreExecutors.directExecutor());

      // ...do some other stuff in this thread...

      // do something with futureDataObj
      futureDataObj.get();
   }
}

据我所知，两者都将通过 executorSvc 运行每个重量级操作，并且都将传播取消和/或执行失败。

似乎 transformAsync(而不是仅将 transform 与 DirectExecutor 以外的执行程序一起使用)的唯一要点是当您使用返回的 API 时ListenableFuture 而不是直接运行该操作。我错过了什么吗？

Answer 1

It seems that the only point of transformAsync (instead of just using transform with an executor other than DirectExecutor) is for when you're working with an API that returns ListenableFuture instead of directly running the operation.

我认为就是这个想法。

但是，我能想到一个让 transformAsync 稍微好一些的小区别:如果你在输出 Future 上调用 cancel(true) , transform 当前 won't interrupt the thread that runs getDetailsFromOtherDB .相反，transformAsync 将(通过调用 ListeningExecutorService 返回的 ListenableFuture 上的 cancel(true))。 transform 应该传播中断，但要做到这一点有一些微妙之处，请参见上面的链接。