java - 如何知道Java中CompletionService中的池方法完成了哪个可调用答案

【问题标题】：How to know which callable finished upon pool method in CompletionService in javajava - 如何知道Java中CompletionService中的池方法完成了哪个可调用
【发布时间】：2018-04-03 18:53:44
【问题描述】：

假设您正在编写如下代码：

CompletionService<T> completion = new ExecutorCompletionService<>(new ExecutorService());

for(Callable callable : callableList) {
    completion.submit(callable);
}

// Do something else

while(true) {
    Future<T> future = completion.poll();

    // At this point, is there a way to find out which callable is returning this future?
}

如果您看到上面的代码，有没有办法找出在我调用completion.poll 时返回的可调用对象并有未来作为回报？

我可以扩展 Callable 以获得某种 ID 来做到这一点，但我想知道 java 本身是否可以通用。

【问题讨论】：

Callable 和 Runnable 一样，是一个非常简单的接口。它不包含标识符数据。

标签： java java.util.concurrent completion-service

【解决方案1】：

TL;DR：虽然可能有一些侵入性的方法（即反射），但ExecutionCompletionService.poll() 方法返回的Future 不会暴露已完成的Callable .

查看ExecutionCompletionService.submit(Callable) 的JDK 9 源代码，提交的Callable 包装在RunnableFuture 中（如果我们查看newTaskFor(Callable)，其实际类型为FutureTask）：

public Future<V> submit(Callable<V> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<V> f = newTaskFor(task);
    executor.execute(new QueueingFuture<V>(f, completionQueue));
    return f;
}

private RunnableFuture<V> newTaskFor(Callable<V> task) {
    if (aes == null)
        return new FutureTask<V>(task);
    else
        return aes.newTaskFor(task);
}

即使调用aes.newTaskFor(task)，其中aes 是AbstractExecutorService，结果也是FutureTask：

public abstract class AbstractExecutorService implements ExecutorService {

    // ...

    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new FutureTask<T>(callable);
    }
}

如果我们查看QueueingFuture 内部类，我们会看到以下定义：

private static class QueueingFuture<V> extends FutureTask<Void> {
    QueueingFuture(RunnableFuture<V> task,
                   BlockingQueue<Future<V>> completionQueue) {
        super(task, null);
        this.task = task;
        this.completionQueue = completionQueue;
    }
    private final Future<V> task;
    private final BlockingQueue<Future<V>> completionQueue;
    protected void done() { completionQueue.add(task); }
}

在submit 方法中传递给新创建的QueueingFuture 的completionQueue 是一个BlockingQueue<Future<T>>，它存储与提供给submit 的已完成Callable 对象对应的Future 对象。换句话说，一旦使用submit 方法提交给ExecutorCompletionService 的Callable 执行完毕，对应于提交的Callable 的Future 将在completionQueue 中排队。 ExecutionCompletionService.poll() 方法只是委托给completionQueue 上的投票：

public Future<V> poll() {
    return completionQueue.poll();
}

因此，我们可以获得对原始Callable 的引用的唯一方法是从ExecutionCompletionService.poll()（您的原始问题）返回的Future 对象中获取它。但是，查看Future 接口，不存在暴露任何Callable 的方法：

public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);
    boolean isCancelled();
    boolean isDone();
    V get() throws InterruptedException, ExecutionException;
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

即使FutureTask（从ExecutionCompletionService.poll()返回的实现类型）也不会暴露其内部Callable：

public class FutureTask<V> implements RunnableFuture<V> {

    private Callable<V> callable;

    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        // ...
    }
}

如果FutureTask 有一些方法可以暴露底层Callable，那么我们也许可以对FutureTask 进行显式强制转换，但既然没有，那么该路由将无法解决手头的问题。

这里的问题是我们正在尝试将同步代码与异步ExecutionCompletionService 结合起来。以同步方式解决此问题的一种方法是在完成时返回 Callable 的 ID，以便在从 ExecutionCompletionService.poll() 返回的 Future 上调用 Future.get() 返回 ID（如您所述）。然后可以将 ID 与其原始Callable 配对。一种异步方法是使用原始Callable 注册一个回调，该回调在Callable 完成时调用。

希望解释比简单的否更有帮助。

【讨论】：