递归并发

Question

我具有以下伪代码功能：

Result calc(Data data) {
  if (data.isFinal()) {
    return new Result(data); // This is the actual lengthy calculation
  } else {
    List<Result> results = new ArrayList<Result>();
    for (int i=0; i<data.numOfSubTasks(); ++i) {
      results.add(calc(data.subTask(i));
    }
    return new Result(results); // merge all results in to a single result
  }
}

我想使用固定数量的线程对其进行并行化。

我的第一次尝试是：

ExecutorService executorService = Executors.newFixedThreadPool(numOfThreads);

Result calc(Data data) {
  if (data.isFinal()) {
    return new Result(data); // This is the actual lengthy calculation
  } else {
    List<Result> results = new ArrayList<Result>();
    List<Callable<Void>> callables = new ArrayList<Callable<Void>>();
    for (int i=0; i<data.numOfSubTasks(); ++i) {
      callables.add(new Callable<Void>() {
        public Void call() {
         results.add(calc(data.subTask(i));
        }
      });
    }
    executorService.invokeAll(callables);  // wait for all sub-tasks to complete
    return new Result(results); // merge all results in to a single result
  }
}

但是，这很快陷入了僵局，因为当顶级递归级别等待所有线程完成时，内部递归级别也等待线程变得可用...

如何有效地并行化程序而不会出现死锁？

Answer 1

使用ThreadPoolExecutor进行具有依赖项的任务时，您的问题是一个常规设计问题。

我看到两个选择：

1）确保以自下而上的顺序提交任务，这样您就永远不会有依赖于尚未启动的任务的正在运行的任务。

2）使用“直接切换”策略（请参阅ThreadPoolExecutor文档）：

ThreadPoolExecutor executor = new ThreadPoolExecutor(poolSize, poolSize, 0, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
executor.setRejectedExecutionHandler(new CallerRunsPolicy());

这个想法是使用同步队列，以便任务永远不会在真实队列中等待。拒绝处理程序负责没有可用线程运行的任务。使用此特定处理程序，提交者线程将运行被拒绝的任务。

此执行程序配置可确保任务永不被拒绝，并且由于任务间的依赖性而不会出现死锁。