java - 实现共享计算的 Future 接口(interface)

Question

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

事实上，我只想放置一个实现 Future<Collection<Integer>> 的类的实例。到 ApplicationScope 对象中，以便任何其他需要结果的线程只需请求 Future来自object并调用方法get()其上，因此使用由另一个线程执行的计算。

我的问题是关于实现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 实例来引用您关心的工作单元。

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

您可能想查看 Guava 提供的并发工具，特别是 ListenableFuture ，它是 Future 的子接口(interface)，提供附加功能。

---

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

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