Future 是否需要在单独的线程中执行计算？

Question

来自documentation

Future 表示异步计算的结果。

这是否意味着调用Future#get 方法的线程不应该是执行计算的线程？那么，如果调用Future#get 的线程尚未开始计算，是否合适？我不确定是否可以称其为异步计算...

Answer 1

术语“异步计算”并不要求计算必须在不同的线程中运行。如果 API 设计者打算这样做，他们已经编写了“在不同线程中运行的计算”。在这里，它只是意味着没有关于计算何时发生的规范。

同样，JRE 提供的现有实现不会在不同的线程中强制执行计算。可能最知名的实现，FutureTask 可以按如下方式使用：

Callable<String> action = new Callable<String>() {
    public String call() {
        return "hello "+Thread.currentThread();
    }
};

FutureTask<String> ft=new FutureTask<>(action);
ft.run();
System.out.println(ft.get());

通常，FutureTask 的实例由 ExecutorService 创建，这将确定何时以及由哪个线程执行计算：

ExecutorService runInPlace=new AbstractExecutorService() {
    public void execute(Runnable command) {
        command.run();
    }
    public void shutdown() {}
    public List<Runnable> shutdownNow() { return Collections.emptyList(); }
    public boolean isShutdown() { return false; }
    public boolean isTerminated() { return false; }
    public boolean awaitTermination(long timeout, TimeUnit unit) { return false; }
};
Future<String> f=runInPlace.submit(action);
System.out.println(ft.get());

请注意，此execute() 实现不违反its contract：

在未来的某个时间执行给定的命令。该命令可以在新线程、池线程或调用线程中执行，由 Executor 实现自行决定。

注意“或在调用线程中”……

另一个实现是ForkJoinTask:

ForkJoinTask<String> fjt=new RecursiveTask<String>() {
    protected String compute() {
        return "hello "+Thread.currentThread();
    }
};
fjt.invoke();
System.out.println(fjt.get());

请注意，虽然这种任务旨在支持拆分为子任务，这些子任务可以由不同的线程执行，但在此有意使用调用线程。如果任务无法拆分，则完全在调用者的线程中运行，这是最有效的解决方案。

---

这些示例都在调用者线程中运行，但它们都不会执行get() 方法中的任务。原则上，这不会违反合同，因为它返回结果值，但是当计算已执行时，您可能会在尝试正确实现 get(long timeout, TimeUnit unit) 时遇到麻烦。相反，拥有一个不工作的cancel() 仍然在合同范围内。

Answer 2

Future.get() 可以使用进行计算的同一线程来完成，但最终结果是不会同时执行任何操作。当第一个执行程序未注释并且第二个执行程序被注释掉时，下面的代码演示了这一点。

可以使用扩展的FutureTask（还有其他选项，如RejectedExecutionHandler）来使用当前线程执行计算，以防计算（下面代码中的Callable）不能同时完成。

至于这是否合适：在某些特殊情况下，这可能是必要的，但这不是它的预期使用方式。对我来说，这似乎是一个过早的优化，只有当我可以看到（测量）遵循预期用法的代码没有提供所需的性能并且专门/优化的代码显示出显着的性能改进（并且满足所需的性能）。

import java.util.concurrent.*;

public class FutureGetSameThread {

    public static void main(String[] args) {

        // Serial executor with a task-queue 
        // ExecutorService executor = Executors.newSingleThreadExecutor();
        // Serial executor without a task-queue 
        ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
        try {
            Callable<Integer> c = new Callable<Integer>() {
                @Override 
                public Integer call() throws Exception {
                    Thread.sleep(400L); // pretend to be busy
                    return 1;
                }
            };
            final Future<Integer> f = executor.submit(c);
            Callable<Integer> c2 = new Callable<Integer>() {
                @Override 
                public Integer call() throws Exception {
                    // wait for the result from the previous callable and add 1
                    return f.get() + 1;
                }
            };
            Future<Integer> f2 = null;
            try {
                f2 = executor.submit(c2);
                System.out.println("Second callable accepted by executor with task queue.");
            } catch (RejectedExecutionException ree) {
                System.out.println("Second callable rejected by executor without task queue.");
                f2 = new FutureTaskUsingCurrentThread<Integer>(c2);
            }
            Integer result = f2.get();
            System.out.println("Result: " + result);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            executor.shutdownNow();
        }
    }

    static class FutureTaskUsingCurrentThread<V> extends FutureTask<V> {

        public FutureTaskUsingCurrentThread(Callable<V> callable) {
            super(callable);
        }

        @Override
        public V get() throws InterruptedException, ExecutionException {
            run();
            return super.get();
        }
    }

}

Answer 3

从技术上讲，java.util.concurrent.Future 只是一个interface。

它是一个可能无法立即可用的值的占位符。可以访问 Future 的方法可以：

• 测试该值是否可用，
• 获取值，或
• “取消”未来。

根据 API 合约，future.get() 操作允许阻塞调用者，直到值可用。

API 合约没有说明值来自哪里，或者为什么它可能不可用，或者什么机制可能会阻止get() 方法的调用者；虽然它说明了“取消”应该看起来如何表现，但它并没有说明“取消”实际上应该做什么。这完全取决于实现。

Future 原本打算由 executorService.submit(callable) 返回。

如果你的方法从一个内置的ExecutorService 实现中获得一个Future，那么当你调用future.get() 时，你实际上是在等待另一个线程产生结果。