Java：ExecutorService线程与CountDownLatch同步导致死锁？

Question

我为编程练习写了一个人生游戏。生成器有 3 种不同的实现。第一个：一个主线程+N个子线程，第二个：SwingWorker + N个子线程，第三个：SwingWorker + ExecutorService。 N 是可用处理器或用户定义的数量。 前两个实现运行良好，有一个或多个线程。 ExecutorServise 的实现在一个线程上运行良好，但在多个线程上锁定。我尝试了所有方法，但我无法得到解决方案。

这里是精细工作实现的代码（第二个）：

    package example.generator;

    import javax.swing.SwingWorker;

    /**
     * AbstractGenerator implementation 2: SwingWorker + sub threads.
     * 
     * @author Dima
     */
    public final class WorldGenerator2 extends AbstractGenerator {


        /**
         * Constructor.
         * @param gamePanel The game panel
         */
        public WorldGenerator2() {
            super();
        }




        /* (non-Javadoc)
         * @see main.generator.AbstractGenerator#startGenerationProcess()
         */
        @Override
        protected void startGenerationProcess() {
            final SwingWorker<Void, Void> worker = this.createWorker();
            worker.execute();
        }




        /**
         * Creates a swing worker for the generation process.
         * @return The swing worker
         */
        private SwingWorker<Void, Void> createWorker() {
            return new SwingWorker<Void, Void>() {

                @Override
                protected Void doInBackground() throws InterruptedException {
                    WorldGenerator2.this.generationProcessing();
                    return null;
                }
            };
        }





        /* (non-Javadoc)
         * @see main.generator.AbstractGenerator#startFirstStep()
         */
        @Override
        public void startFirstStep() throws InterruptedException {
            this.getQueue().addAll(this.getLivingCells());
            for (int i = 0; i < this.getCoresToUse(); i++) {
                final Thread thread = new Thread() {
                    @Override
                    public void run() {
                        WorldGenerator2.this.fistStepProcessing();
                    }
                };
                thread.start();
                thread.join();
            }
        }




        /* (non-Javadoc)
         * @see main.generator.AbstractGenerator#startSecondStep()
         */
        @Override
        protected void startSecondStep() throws InterruptedException {
            this.getQueue().addAll(this.getCellsToCheck());
            for (int i = 0; i < this.getCoresToUse(); i++) {
                final Thread thread = new Thread() {

                    @Override
                    public void run() {
                        WorldGenerator2.this.secondStepProcessing();
                    }
                };
                thread.start();
                thread.join();
            }
        }

    }

这里是执行器服务不工作的代码：

    package example.generator;

    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;

    import javax.swing.SwingWorker;

    /**
     * AbstractGenerator implementation 3: SwingWorker + ExecutorService.
     * 
     * @author Dima
     */
    public final class WorldGenerator3 extends AbstractGenerator {


        private CountDownLatch  countDownLatch;
        private ExecutorService executor;


        /**
         * Constructor.
         * @param gamePanel The game panel
         */
        public WorldGenerator3() {
            super();
        }




        /* (non-Javadoc)
         * @see main.generator.AbstractGenerator#startGenerationProcess()
         */
        @Override
        protected void startGenerationProcess() {
            this.executor = Executors.newFixedThreadPool(this.getCoresToUse());
            final SwingWorker<Void, Void> worker = this.createWorker();
            worker.execute();
        }




        /**
         * Creates a swing worker for the generation process.
         * @return The swing worker
         */
        private SwingWorker<Void, Void> createWorker() {
            return new SwingWorker<Void, Void>() {

                @Override
                protected Void doInBackground() throws InterruptedException {
                    WorldGenerator3.this.generationProcessing();
                    return null;
                }
            };
        }




        /* (non-Javadoc)
         * @see main.generator.AbstractGenerator#startFirstStep()
         */
        @Override
        public void startFirstStep() throws InterruptedException {
            this.getQueue().addAll(this.getLivingCells());
            this.countDownLatch = new CountDownLatch(this.getCoresToUse());
            for (int i = 0; i < this.getCoresToUse(); i++) {    
                this.executor.execute(new Runnable() {  
                    @Override
                    public void run() {
                        WorldGenerator3.this.fistStepProcessing();
                        WorldGenerator3.this.countDownLatch.countDown();
                    }
                });
            }
            this.countDownLatch.await();

        }




        /* (non-Javadoc)
         * @see main.generator.AbstractGenerator#startSecondStep()
         */
        @Override
        protected void startSecondStep() throws InterruptedException {
            this.getQueue().addAll(this.getCellsToCheck());
            this.countDownLatch = new CountDownLatch(this.getCoresToUse());
            for (int i = 0; i < this.getCoresToUse(); i++) {
                this.executor.execute(new Runnable() {
                    @Override
                    public void run() {
                        WorldGenerator3.this.secondStepProcessing();
                        WorldGenerator3.this.countDownLatch.countDown();
                    }
                });
            }
            this.countDownLatch.await();

        }
    }

您可以在这里下载我的应用程序示例，带有一个小型启动器。它仅在控制台上打印迭代的结果：Link

---

现在我的代码如下所示：

/* (non-Javadoc)
 * @see main.generator.AbstractGenerator#startFirstStep()
 */
@Override
public void startFirstStep() throws InterruptedException {

    this.getQueue().addAll(this.getLivingCells());      

    final ArrayList<Callable<Void>> list = new ArrayList<Callable<Void>>(this.getCoresToUse());

    for (int i = 0; i < this.getCoresToUse(); i++) {

        list.add(new Callable<Void>() {

                @Override
                public Void call() throws Exception {
                    WorldGenerator3.this.fistStepProcessing();
                    return null;
                }
            }
        );          
    }

    this.executor.invokeAll(list);
}

但这又是同样的问题。如果我用一个核心（线程）运行它，就没有问题。如果我将核心数设置为一个以上，它会锁定。在我的第一个问题中，有一个示例链接，您可以运行该示例（在 Eclipse 中）。也许我忽略了前面代码中的某些内容。

Answer 1

我发现您对 Executors 设施的使用有点奇怪...

即这个想法是让 Executor 有一个线程池，其大小通常与你的 CPU 支持的内核数有关。 然后，您将任意数量的并行任务提交给 Executor，让它决定何时以及在其池中的哪个可用线程上执行什么。

至于 CountDownLatch... 为什么不使用ExecutorService.invokeAll？此方法将阻塞，直到所有提交的任务都完成或达到超时。因此，它将代表您计算剩余的工作。 或者CompletionService，如果您想在任务结果可用时立即使用它，即“不等待所有任务首先完成，则将新异步任务的产生与已完成任务结果的消耗分离”。

类似

    private static final int WORKER_THREAD_COUNT_DEFAULT = Runtime.getRuntime().availableProcessors() * 2;

    ExecutorService executor = Executors.newFixedThreadPool(WORKER_THREAD_COUNT);

    // your tasks may or may not return result so consuming invokeAll return value may not be necessary in your case 
    List<Future<T>> futuresResult = executor.invokeAll(tasksToRunInParallel, EXECUTE_TIMEOUT,
                TimeUnit.SECONDS);

Answer 2

在所有变体中，您都在串行而不是并行执行线程，因为您在 for 循环中 join 和 await。这意味着在刚刚启动的线程完成之前，for循环不能继续进行下一次迭代。这相当于在任何给定时间只有一个线程存在——要么是主线程，要么是在当前循环迭代中创建的一个线程。如果你想加入多个线程，你必须收集它们的引用，然后，在你启动它们的循环之外，进入另一个你加入每个线程的循环。

至于在Executors 变体中使用CountDownLatch，这里所说的线程用于闩锁：不要使用实例变量；使用一个本地列表来收集所有锁存器并在单独的循环中等待它们。

但是，您首先不应该真正使用CountDownLatch：您应该将所有并行任务放在Callables 的列表中并用它调用ExecutorService.invokeAll。它会自动阻塞，直到所有任务完成。