Java从ExecutorService设置回调

Question

我有一个fixedThreadPool,我用它来运行一堆工作线程来实现具有许多组件的任务的并行执行.

当所有线程都完成后,我使用方法(getResult)检索它们的结果(它们非常大)并将它们写入文件.

最终,为了节省内存并能够看到中间结果,我希望每个线程在完成执行后立即将其结果写入文件,然后释放其内存.

通常,我会在run()方法的末尾添加代码.但是,此类中的某些其他对象也会调用这些线程,但不要让它们将结果写入文件 - 而是使用其结果执行其他计算,最终将其写入文件.

所以,我想知道是否可以使用ExecutorService将回调函数附加到线程完成事件.这样,我可以立即检索其结果并释放该场景中的内存,但在其他场景中使用这些线程时不会破坏代码.

这样的事情可能吗？

Answer 1

ExecutorService#submitreturn FutureTask<T>可以帮助您检索结果,该ExecutorService#get方法将阻止执行,直到计算未完成.示例 -

ExecutorService executor = Executors.newFixedThreadPool(10);
Future<Long> future = executor.submit(new Callable<Long>(){
       @Override
       public Long call() throws Exception {
           long sum = 0;
           for (long i = 0; i <= 10000000l; i++) {
               sum += i;
           }
           return sum;
       }
});
Long result = future.get();
System.out.println(result);

Answer 2

如果使用Google Guava是一个选项,您可以通过以下方式使用ListenableFuture界面:

1. 转换一个ExecutorService到ListeningExecutorService通过MoreExecutors.listeningDecorator(existingExecutorService)
2. 已缩小的submit(Callable<V>)方法ListeningExecutorService返回a ListenableFuture,这是一个子接口Future.
3. ListenableFuture有一个addListener()方法,所以你可以注册一个回调,以便在未来完成时运行.

Answer 3

您可以在 Java 8+ 中添加一个线程在 Java 8+ 中返回时的回调，CompletableFuture如下所示，其中t是长时间运行的计算结果，

CompletableFuture.supplyAsync(() -> {
    T t = new T();
    // do something
    return t;
}).thenApply(t -> {
    // process t
});

如果你只想在 Java 7 中使用回调，你可以这样做，

int x = 10;
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(x);
Future<T> result = fixedThreadPool.submit(() -> {
    // do calculation
    return T;
});
fixedThreadPool.submit(() -> {
    long minutesToWait = 5;
    T t = null;
    try {
        t = result.get(minutesToWait, TimeUnit.MINUTES);
    } catch (InterruptedException | ExecutionException | TimeoutException e) {
        LOGGER.error(e);
    }
    if (t != null) {
        // process t
    }
});

Answer 4

因此，我想知道是否可以将回调函数附加到使用 ExecutorService 完成线程的事件。

不是直接的，不，但是有几种方法可以实现这一点。我想到的最简单的方法是将您的内容包装Runnable在另一个Runnable可以收获结果的内容中。

所以你会做类似的事情：

threadPool.submit(new ResultPrinter(myRunnable));
...

private static class ResultPrinter implements Runnable {
    private final MyRunnable myRunnable;
    public ResultPrinter(MyRunnable myRunnable) {
        this.myRunnable = myRunnable;
    }
    public void run() {
        myRunnable.run();
        Results results = myRunnable.getResults();
        // print results;
    }
}