java - 取消 Future 并仍然检索其返回值

Question

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

我的想法是拥有一个带有单线程的 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 更改为推送已收集的信息而不是返回它。

换句话说:不是使用 Future 在某个时刻获取所有数据；为什么不直接在到达时将所有数据发送到“安全接收器”？你的方法是民意调查；但显然:轮询和中断并不是齐头并进的!

编辑:现在您可以像这样创建记录器:

 public Recorder(DataSource source)

因此您构建了从源接收“数据”的功能。一个简单的解决方案可能是

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

还有。

并且无需在 Runable 中保留本地 dataList ...您只需附加到提供给记录器的 sink 即可。

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