为共享计算实现 Future 接口

Question

我正在实现Future<Collection<Integer>> 接口，以便在应用程序中的所有线程之间共享一些批量计算的结果。

事实上，我打算将实现Future<Collection<Integer>> 的类的实例放入ApplicationScope 对象中，以便任何其他需要结果的线程只需从object 请求Future 并调用方法@ 987654326@ 就可以了，因此使用了另一个线程执行的计算。

我的问题是关于实现cancel 方法。现在，我会写这样的东西：

public class CustomerFutureImpl implements Future<Collection<Integer>>{

    private Thread computationThread;
    private boolean started;
    private boolean cancelled;
    private Collection<Integer> computationResult;

    private boolean cancel(boolean mayInterruptIfRunning){
        if( computationResult != null )
            return false;
        if( !started ){
            cancelled = true;
            return true;
        } else {
            if(mayInterruptIfRunning)
                 computationThread.interrupt();
        }
    }

    //The rest of the methods
}

但是方法实现不满足Future的文档，因为我们需要在任何等待结果的线程中抛出CancellationException（已调用get()方法）。

我应该添加另一个像private Collection<Thread> waitingForTheResultThreads; 这样的字段，然后从Collection 中断每个线程，捕获中断的异常，然后throw new CancellationException()？

问题是这样的解决方案对我来说似乎有点奇怪……我不确定。

Answer 1

通常你应该避免直接实现Future。并发代码非常很难正确处理，分布式执行框架 - 特别是 ExecutorService - 将提供引用您关心的工作单元的 Future 实例。

您可能已经知道并且正在有意创建一个新的类似服务，但我认为对于广大大多数用例来说，重要的是，您不需要定义自己的 @ 987654330@实现。

您可能想查看 Guava 提供的并发工具，特别是 ListenableFuture，它是 Future 的子接口，提供了额外的功能。

---

假设您确实想要定义自定义 Future 类型，请使用 Guava 的 AbstractFuture 实现作为起点，这样您就不必重新设计遇到的复杂细节。

对于您的具体问题，如果您查看implementation of AbstractFuture.get()，您会发现它是使用while 循环实现的，该循环查找value 变为非空，此时它调用getDoneValue()它要么返回值，要么引发CancellationException。所以本质上，每个在调用Future.get() 时阻塞的线程都会不时地轮询Future.value 字段，如果检测到Future 已被取消，则引发CancellationException。无需跟踪Collection<Thread> 或任何类似的东西，因为每个线程都可以独立检查Future 的状态，并根据需要返回或抛出。