有没有一种简单的方法可以将 Future<Future<T>> 变成 Future<T>？

Question

我有一些代码可以将请求提交给另一个线程，该线程可能会也可能不会将该请求提交给另一个线程。这会产生Future<Future<T>> 的返回类型。是否有一些不令人发指的方法可以立即将其变成等待整个未来链完成的Future<T>？

我已经在使用 Guava 库来处理其他有趣的并发东西，并作为 Google Collections 的替代品，它运行良好，但我似乎找不到适合这种情况的东西。

Answer 1

使用番石榴库的另一种可能的实现方式要简单得多。

import java.util.concurrent.*;
import com.google.common.util.concurrent.*;
import com.google.common.base.*;

public class FFutures {
  public <T> Future<T> flatten(Future<Future<T>> future) {
    return Futures.chain(Futures.makeListenable(future), new Function<Future<T>, ListenableFuture<T>>() {
      public ListenableFuture<T> apply(Future<T> f) {
        return Futures.makeListenable(f);
      }
    });
  }
}

Answer 2

Guava 13.0 添加了Futures.dereference 来执行此操作。它需要ListenableFuture<ListenableFuture>，而不是普通的Future<Future>。 （在普通的 Future 上操作需要一个 makeListenable 调用，每个调用都需要一个用于任务生命周期的专用线程（方法的新名称JdkFutureAdapters.listenInPoolThread 更清楚地说明了这一点。）

Answer 3

我认为这是执行 Future 合约所能做的最好的事情。我采取了尽可能不聪明的策略，以确保它符合合同。尤其是 get with timeout 的实现。

import java.util.concurrent.*;

public class Futures {
  public <T> Future<T> flatten(Future<Future<T>> future) {
    return new FlattenedFuture<T>(future);
  }

  private static class FlattenedFuture<T> implements Future<T> {
    private final Future<Future<T>> future;

    public FlattenedFuture(Future<Future<T>> future) {
      this.future = future;
    }

    public boolean cancel(boolean mayInterruptIfRunning) {
      if (!future.isDone()) {
        return future.cancel(mayInterruptIfRunning);
      } else {
        while (true) {
          try {
            return future.get().cancel(mayInterruptIfRunning);
          } catch (CancellationException ce) {
            return true;
          } catch (ExecutionException ee) {
            return false;
          } catch (InterruptedException ie) {
            // pass
          }
        }
      }
    }

    public T get() throws InterruptedException, 
                          CancellationException, 
                          ExecutionException 
    {
      return future.get().get();
    }

    public T get(long timeout, TimeUnit unit) throws InterruptedException, 
                                                     CancellationException, 
                                                     ExecutionException, 
                                                     TimeoutException 
    {
      if (future.isDone()) {
        return future.get().get(timeout, unit);
      } else {
        return future.get(timeout, unit).get(0, TimeUnit.SECONDS);
      }
    }

    public boolean isCancelled() {
      while (true) {
        try {
          return future.isCancelled() || future.get().isCancelled();
        } catch (CancellationException ce) {
          return true;
        } catch (ExecutionException ee) {
          return false;
        } catch (InterruptedException ie) {
          // pass
        }
      }
    }

    public boolean isDone() {
      return future.isDone() && innerIsDone();
    }

    private boolean innerIsDone() {
      while (true) {
        try {
          return future.get().isDone();
        } catch (CancellationException ce) {
          return true;
        } catch (ExecutionException ee) {
          return true;
        } catch (InterruptedException ie) {
          // pass
        }
      }
    }
  }
}

Answer 4

你可以像这样创建一个类：

public class UnwrapFuture<T> implements Future<T> {
    Future<Future<T>> wrappedFuture;

    public UnwrapFuture(Future<Future<T>> wrappedFuture) {
        this.wrappedFuture = wrappedFuture;
    }

    public boolean cancel(boolean mayInterruptIfRunning) {
        try {
            return wrappedFuture.get().cancel(mayInterruptIfRunning);
        } catch (InterruptedException e) {
            //todo: do something
        } catch (ExecutionException e) {
            //todo: do something
        }
    }
    ...
}

您必须处理 get() 可以引发但其他方法不能引发的异常。

Answer 5

这是我第一次尝试它，但我确信它有很多问题。我很乐意将其替换为 Futures.compress(f) 之类的内容。

public class CompressedFuture<T> implements Future<T> {
    private final Future<Future<T>> delegate;

    public CompressedFuture(Future<Future<T>> delegate) {
        this.delegate = delegate;
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        if (delegate.isDone()) {
            return delegate.cancel(mayInterruptIfRunning);
        }
        try {
            return delegate.get().cancel(mayInterruptIfRunning);
        } catch (InterruptedException e) {
            throw new RuntimeException("Error fetching a finished future", e);
        } catch (ExecutionException e) {
            throw new RuntimeException("Error fetching a finished future", e);
        }
    }

    @Override
    public T get() throws InterruptedException, ExecutionException {
        return delegate.get().get();
    }

    @Override
    public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        long endTime = System.currentTimeMillis() + unit.toMillis(timeout);
        Future<T> next = delegate.get(timeout, unit);
        return next.get(endTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }

    @Override
    public boolean isCancelled() {
        if (!delegate.isDone()) {
            return delegate.isCancelled();
        }
        try {
            return delegate.get().isCancelled();
        } catch (InterruptedException e) {
            throw new RuntimeException("Error fetching a finished future", e);
        } catch (ExecutionException e) {
            throw new RuntimeException("Error fetching a finished future", e);
        }
    }

    @Override
    public boolean isDone() {
        if (!delegate.isDone()) {
            return false;
        }
        try {
            return delegate.get().isDone();
        } catch (InterruptedException e) {
            throw new RuntimeException("Error fetching a finished future", e);
        } catch (ExecutionException e) {
            throw new RuntimeException("Error fetching a finished future", e);
        }
    }
}