java - 并行运行操作，同时保留有序的增量输出

Question

我有一个缓慢的 CPU 密集型操作:doWork(int x)，它使用具有不同值的单个整数参数调用，如下所示:

static String doWork(int x) {
  // do work that depends on i, takes ~60 seconds
  ...
}

public static void main(String args[]) {
  for (int i = 1; i < 100; i++) {
    System.println(doWork(i));
  }
}

当每个 doWork() 调用完成时，结果会输出到控制台。我想将其并行化 - 所有 doWork() 调用都是独立的，不会改变任何共享状态。现在，我可以用老方法来做，搞乱 ExecutorSevice 和 Future.get() 等等，但我想用流更干净地做¹.

所以像这样的东西看起来几乎可以工作:

public static void main(String args[]) {
    IntStream.rangeClosed(1, 100).parallel()
        .forEach(i -> System.out.println(doWork(i)));
}

...但问题是我想保留控制台上的输出顺序(doWork(1) 行应该排在第一位，依此类推)。我不能使用 forEachOrdered()，因为它会序列化整个操作:只会使用一个线程。问题的根源在于 forEachOrdered 提供了一种过于强大的保证:一次在一个元素上顺序调用消费者方法。我希望并行调用消费者，但按顺序输出。

所以我可能应该看看 map -> collect type idiom，我将每个 doWork() 调用的输出收集到一个字符串中并打印一次:

public static void main(String[] args) {
    System.out.println(IntStream.rangeClosed(1, 100).parallel()
        .mapToObj(Main::doWork).collect(Collectors.joining("\n")));
}

差不多了! collect() 方法保持遇到顺序，所以我的元素是有序的。现在的问题是没有增量输出——整个作业必须在任何输出发生之前完成。我真的很想保留更新在控制台上运行的行为。

我想我想要某种有序消费终端操作，它不会强制整个管道被排序。基本上它会像普通收集器一样在内部收集结果，但是当收集到当前“最左边”的元素时，它会把它传递给消费者——所以消费者会看到一系列有序的元素，但一切仍然是并行发生的。

那里有什么东西吗？似乎不可能在现有的 Collector 接口(interface)上构建它，因为它没有为您提供确定元素顺序的方法。

---

¹ ...也许更有效，因为 fork/join 是在幕后使用的，所以也许我可以利用该框架内置的一些启发式方法？

Answer 1

你很接近。只需组合 map 和 forEachOrdered 解决方案:

IntStream.rangeClosed(1, 100)
         .parallel()
         .mapToObj(Main::doWork)
         .forEachOrdered(System.out::println);