取消一个 Future 并仍然检索它的返回值

Question

我有一个记录器类，它每 n 毫秒从数据源轮询一次值。整个 this 应该是异步运行并返回一个包含记录数据的集合。

我的想法是拥有一个带有单线程的 ExecutorService 和一个执行数据轮询的 Callable。可调用对象应该被中断，因此调用者不必等待任何Thread.sleep 调用完成。为了打断他 Callable，我在 ExecutorService 返回的 Future 上调用 Future.cancel(true)：

class Recorder {
    private DataSource source;
    private ExecutorService executor;
    private Future<List<MyData>> dataFuture;

    private class FixedPollRateCallable {
        private DataSource source;
        private long pollrate;

        public FixedPollrateCallable(DataSource source, long pollrate) {
            this.source = source;
            this.pollrate = pollrate;
        }

        public List<MyData> call() throws Exception {
            List<MyData> dataList = new ArrayList<>();
            while(!Thread.currentThread().isInterrupted()) {
                dataList.add(source.getData());
                try {
                    Thread.sleep(pollrate);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
            return dataList;
        }
    }

    public Recorder(DataSource source) {
        this.source = source;
        this.executor = Executors.newSingleThreadExecutor();
    }

    public void startRecording(long pollrate) {
        if(dataFuture != null && !dataFuture.isDone())
            throw new RecordingException();

        dataFuture = executor.submit(new FixedPollRateCallable(source, pollrate);
    }

    public void stopRecording() {
        dataFuture.cancel(true);
    }

    public List<MyData> getRecordedData() throws InterruptedException, ExecutionException {
        return dataFuture.get();
    }
}

但是当这样调用整个事情时：

recorder.startRecording();
recorder.stopRecording();
recorder.getRecordedData();

我在recorder.getRecordedData() 通话中收到java.util.concurrent.CancellationException。

我猜cancel(true)除了中断线程外，还设置了一个取消标志，所以对Future.get()的调用总是会失败。

有什么办法可以解决这个问题，或者你知道一个更好的替代方案，可以让我中断线程，突破Thread.sleep() 并且仍然能够返回一个值？

Answer 1

好吧，你要求的东西在概念上没有意义：

让我们引用 get() 的 javadoc：

抛出：CancellationException - 如果计算被取消

对于cancel()，它说：

此方法返回后，后续对 isDone() 的调用将始终返回 true。如果此方法返回 true，则后续调用 isCancelled() 将始终返回 true。

换句话说：您可能需要一些其他类型的机制。例如，您可以将 Runable 更改为 push 已收集的信息，而不是返回它。

换句话说：不是使用 Future 在某个时间点获取所有数据；为什么不直接在到达时将所有数据发送到“安全接收器”？你的方法是投票；但显然：轮询和中断不能齐头并进！

编辑：现在你像这样创建你的记录器：

 public Recorder(DataSource source)

所以你构建接收来自source 的“数据”。一个简单的解决方案可能是拥有

 public Recorder(DataSource source, List<MyData> sink)

也是。

而不是在您的 Runable 中保留本地 dataList ...您只需附加到提供给记录器的 sink。

我使用“安全”这个词只是指例如当多个线程开始更新同一个列表时需要注意的事实。