队列并行处理的停止条件，其中任务可能产生更多任务

Question

我正在使用PriorityBlockingQueue<T> q 来处理任务的并行处理。队列由一些任务初始化，每个任务的处理可能会产生更多的任务，这些任务将被添加到队列中。我想并行处理任务，并在处理完所有任务后停止。当然，如果任务仍在被其他线程处理，队列可能会在我们完成之前暂时变空。

我的问题是：在 Java 中，什么是好的（当然是正确的，但也是优雅、惯用的，没有不必要的锁定或等待队列为空时）的方法？

注意事项：

1. 我正在使用优先级队列，但任务可以按任何顺序处理（大致按照优先级指定的顺序处理任务有一些好处 - 但我认为忽略这一位是安全的)。

2. This answer（“使用任务并行库”）似乎解决了 C# 的问题，但 it seems 这对于 Java 不存在。 This 本质上是同一个问题；它似乎没有一个完全令人满意的答案......

3. 每个任务的处理都相当冗长，所以如果它可以使代码更优雅，那么任务管理的开销会增加一点（这也是我很高兴在其中使用单线程的原因）负责从队列中轮询任务并将它们分配给工作人员）

示例： 作为一个粗略的近似，我正在尝试使用并行 BFS 来查找动态生成的树的深度。您可以将其视为在玩游戏时寻找一种能够最大化您的奖励的策略，您的每一步都可以获得积分。每个状态都是要处理的任务。您从初始（根）状态开始，计算所有移动（此计算很长，可能会产生数千个移动），并将这些移动达到的状态添加为要探索的任务。

Answer 1

我意识到一个解决方案可能应该允许所有线程递归地提交新任务，这让我找到了this answer。

这是该答案的完整版本，它处理二叉树的遍历，通过从某个字符串开始并将其大致减半而获得。 为了支持优先级，可以简单地修改 MyTask.run() 以从一些辅助 PriorityBlockingQueue 中弹出字符串，我决定省略它，因为它只会混淆解决方案的本质。

        int nProcessors = Runtime.getRuntime().availableProcessors();
        System.out.println("number of processors found " + nProcessors);
        final ExecutorService taskExecutor = Executors.newFixedThreadPool(nProcessors);

        final Phaser phaser = new Phaser();
        int phase = phaser.getPhase();

        class MyTask implements Runnable {
            private String input;

            public MyTask(String input) {
//        this.es = es;
                this.input = input;
            }

            public void run() {
                System.out.println("running on \"" + this.input + "\" invoked in thread " + Thread.currentThread().getName());

                int l = this.input.length();
                if (! ((l == 0) || (l == 1))) {
                    String beginning = this.input.substring(0, l / 2);
                    String ending = this.input.substring(l / 2);
                    phaser.register();
                    phaser.register();
                    taskExecutor.execute(new MyTask(beginning));
                    taskExecutor.execute(new MyTask(ending));
                }
                phaser.arrive();
            }
        }

        phaser.register();
        taskExecutor.execute(new MyTask("a lengthy string to be processed goes here"));
        System.out.println("MAIN: initial task submitted at " + Thread.currentThread().getName() + " awaiting phaseAdvance");
        phaser.awaitAdvance(phase);
        System.out.println("MAIN: all tasks arrived, ok to shutdown?");
        try {
            taskExecutor.shutdown();
            System.out.println("MAIN: sutting down awaiting termination...");
            taskExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
            System.out.println("MAIN: await termination finished.");
        } catch (InterruptedException e) {
            System.out.println("caught an interrupted excepion " + e);
        }

输出是

number of processors found 4
MAIN: initial task submitted at main awaiting phaseAdvance
running on "a lengthy string to be processed goes here" invoked in thread pool-1-thread-1
running on "a lengthy string to b" invoked in thread pool-1-thread-2
running on "e processed goes here" invoked in thread pool-1-thread-3
running on "e processe" invoked in thread pool-1-thread-3
running on "a lengthy " invoked in thread pool-1-thread-4
running on "string to b" invoked in thread pool-1-thread-3
running on "d goes here" invoked in thread pool-1-thread-3
running on "e pro" invoked in thread pool-1-thread-3
running on "cesse" invoked in thread pool-1-thread-3
.
.
.
running on "b" invoked in thread pool-1-thread-2
running on " " invoked in thread pool-1-thread-1
running on "e" invoked in thread pool-1-thread-4
running on "i" invoked in thread pool-1-thread-1
running on "h" invoked in thread pool-1-thread-2
running on " " invoked in thread pool-1-thread-3
running on "n" invoked in thread pool-1-thread-4
MAIN: all tasks arrived, ok to shutdown?
MAIN: sutting down awaiting termination...
MAIN: await termination finished.