在 Callbable 与 Runnable 中使用“ExecutorCompletionService”的问题

Question

我一直在使用 ExecutorCompletionService 中的示例代码，并将以下示例代码放在一起。 solve() 中的代码按预期工作并打印出
1
2
3
4
5
solve2() 中的代码不打印任何内容，实际上永远不会退出。 ecs 是在提交作业到 ExecutionService 之前还是之后构建都没有关系。

有没有办法将 CompletionService 构造与 FutureTasks 一起使用？我已经重写了我的生产代码以直接获取（） FutureTask 的结果，而不是尝试从 ExecutorCompletionService 获取（）它们，但它（当前）导致了一些看起来很乱的东西。简而言之，下面的solve2有什么问题？谢谢。

import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;

public class sample {
public static class stringCallable implements Callable<String>{
    String mstring;

    stringCallable(String s) {mstring = s;}
    @Override
    public String call() throws Exception {
        // TODO Auto-generated method stub
        return mstring;
    }
};

public static void main(String[] args) {
    // TODO Auto-generated method stub
    ArrayList<Callable<String>> list = new ArrayList<Callable<String>>();
    ExecutorService es = Executors.newFixedThreadPool(1);
    Executor e = Executors.newSingleThreadExecutor();
    list.add(new stringCallable("1"));
    list.add(new stringCallable("2"));
    list.add(new stringCallable("3"));
    list.add(new stringCallable("4"));
    list.add(new stringCallable("5"));

    try {
        solve(e, list);
    } catch (InterruptedException e1) {
        // TODO Auto-generated catch block
        e1.printStackTrace();
    } catch (ExecutionException e1) {
        // TODO Auto-generated catch block
        e1.printStackTrace();
    }
    System.out.println ("Starting Solver 2");

    try {
        solve2(es, list);
    } catch (InterruptedException e1) {
        // TODO Auto-generated catch block
        e1.printStackTrace();
    } catch (ExecutionException e1) {
        // TODO Auto-generated catch block
        e1.printStackTrace();
    }
}

static void solve(Executor e, Collection<Callable<String>> solvers)throws InterruptedException, ExecutionException {
    CompletionService<String> ecs = new ExecutorCompletionService<String>(e);
    for (Callable<String> s : solvers)
     ecs.submit(s);
    int n = solvers.size();
    for (int i = 0; i < n; ++i) {
        String r = ecs.take().get();
        if (r != null)
            use(r);
    }
}

static void solve2(ExecutorService e, Collection<Callable<String>> solvers)throws InterruptedException, ExecutionException {
    for (Callable<String> s : solvers){
        FutureTask<String> f = new FutureTask<String>(s);
          e.submit(f);
    }
    CompletionService<String> ecs = new ExecutorCompletionService<String>(e);
    int n = solvers.size();
    for (int i = 0; i < n; ++i) {
        String r = ecs.take().get();
        if (r != null)
            use(r);
    }
}

private static void use(String r) {
    System.out.println (r);
}

}

Answer 1

在solve2 中，当您使用现有ExecutorService 创建ExecutorCompletionService 时，包装器会忽略它提交的任务，因为它使用单独的LinkedBlockingQueue。提交的任务不会被继承。因此，当您执行 ecs.take().get(); 时，您的代码会阻塞，因为 ExecutorCompletionService 本身没有任何提交的任务。

此外，您无需专门创建 FutureTask 即可提交给ExecutorCompletionService。这些未来任务已经在内部为您创建。这就是为什么您在调用ecs.take(); 时会得到Future<String>。

鉴于此，您的 solve2 函数完全没用。您已经在solve1 中正确执行此操作。

Answer 2

这就是我将如何实现它：

static void solve2(ExecutorService e, Collection<Callable<String>> solvers)throws InterruptedException, ExecutionException {
    CompletionService<String> ecs = new ExecutorCompletionService<String>(e);
    for (Callable<String> s : solvers){
        ecs.submit(s);
    }
    int n = solvers.size();
    for (int i = 0; i < n; ++i) {
        String r = ecs.take().get();
        if (r != null)
            use(r);
    }
}

ExecutorCompletionService 只是 ExecutorService 的一个包装器，但您必须将您的可调用对象提交给 ECS，因为 ECS 将获取可调用对象的结果，并将其放入队列中。然后可以通过 take() 或 poll() 获得该结果。 如果直接在 ExecutorService 上提交 callable，ECS 无法知道它的完成情况。 如果您查看 ECS 的 javadoc，它会说完全相同的内容 + 很好的示例（甚至更好的解释）。我建议你也看看源代码 java.util.concurrent.ExecutorCompletionService