将一个Spliterator分成N个Spliterator

Question

在分析一个同事几年前的 Java 7 代码时，我发现他实现了一种用于遍历数据（可能是并行的）的工具。他称之为 Range，并扩展了 Iterator 接口。它的一些新方法有些奇怪地熟悉：

• int size() 可以给出范围的确切大小；
• Range split() 可以将范围分成两个部分，最好是相似的大小（修改当前范围并创建一个新范围）；
• Range[] split(int n) 可以将范围分成 N 个子范围，可能会尝试使它们尽可能均匀。
• 尽管 Iterator 中有 void remove()，但其子类型只是会抛出 UnsupportedOperationException。

我对自己说，这绝对可以用（subsized）spliterator 实现，从而利用流 API。然而，Java 8 中的 spliterator 没有提供将其分成 n 部分的简单方法。

通过递归函数，我们可以将其分成 n=2^L 部分，其中 L 是递归级别，但当 n 不是两的幂时，这种方法并不那么简单，更不用说保留一个工具函数的效果了。

也可以说要避免与 spliterator 搞在一起，让流在实际处理过程中进行 fork 所引发的分裂，但 ForkJoin 策略可能不太适合此任务，并且不能保证会使用我们专门为该工作分配的线程数。事实上，有可能在少量元素上执行繁重任务。

问题概括为：如果有一个至少具备 SIZED 和 SUBSIZED 特征的 spliterator，如何将其分成精确数量的 spliterator？

- E_net4

Answer 1

做到这一点的方法并不是很巧妙，而且可以与现有的 F/J 支持框架进行互操作，就是编写一个 spliterator 包装器，该包装器使用自己的分割策略。你失去的是利用本地随机访问结构的能力；你只能通过迭代内部 splitetator 的数据集来拆分。

你可以参考我的 早先的回答，我在那里提供了将数据分成预定大小批次的代码；只需适当的构造函数调用即可轻松地将其适应到您的情况。

- Marko Topolnik

---

说句奇怪的话，我在阅读了您的博客文章后不久就提出了这个问题。虽然它让我离解决方案更近了一步，但仍然着重于修复批次大小而不是处理核心数。尽管如此，我现在会尝试采纳您的建议，并在有结果时向您回报。 - E net4

这是我的看法，但我认为同样的方法非常容易适应基于核心计数驱动的分裂策略。在我看来，你真正的问题将是这种方法的顺序性质。一个SIZED spliterator最有可能由一个数组支持，利用这个事实的实现将会更加显著（O（1）分裂代替O（n），在空间和时间上都是如此）。 - Marko Topolnik

你说得对！事实上，我们目前的实现要么由 ArrayList 支持，要么由 Lucene 的 IndexReader 支持。提前了解这些细节肯定会提高性能，但是在这里拥有一个通用的解决方案也是不错的。 - E net4

我个人有使用IndexReader支持的并行化经验（这实际上也是我编写这些类的原因）。由于Lucene操作是重量级的（100微秒的查找并不罕见），即使在顺序模式下并行化也是值得的。请注意，在这种情况下，将核心数与批次数精确匹配并不那么重要。给每个批次足够的工作量（例如10毫秒），协调/切换开销就会消失。 - Marko Topolnik

Answer 2

“…但ForkJoin策略可能不太适合这个任务”，听起来像是过早优化。你想要手动实现复杂的事情，因为现有的策略可能不足以胜任...

“...并不能保证它将使用我们专门希望用于该工作的线程数。”确实没有保证，但当前的流实现使用常见的Fork/Join池，可以配置为所需数量的线程。将指定数量的线程分配给任务正是您要求的策略。

“实际上，可能会出现在少量元素上执行重任务的情况。”我没有看到与F/J框架工作方式的矛盾。它将尝试拆分，直到达到所需的并行性为止。如果这意味着每个线程只处理一个项目，那么它就会这样做。

在这一点上，值得注意的是，默认并行性与核心数匹配，对于不涉及阻塞的任何计算任务都足够了，无论处理单个项目需要多长时间。只要每个线程都有其工作负载，就没有超出实际硬件执行单元数量的加速可能。

换句话说，F/J框架实现了您想要实现自己的策略（或比您实现的策略更优），这让我们回到第一个点，过早优化。

- Holger

---

如果您的任务少于线程数，您必须拆分任务本身以获得更多的并行性，这是没有人可以为您完成的，因为只有您知道实际任务及其如何拆分（如果可能的话）。我不明白收集项目并传递到另一个池中如何帮助您，在监视工具中可能会创建看起来很奇怪的线程实例数量，但对于并行性而言没有任何好处，因为第二个池的线程都必须等待第一个池执行的收集完成，因此它们不能同时运行。 - Holger

@Marko Topolnik：你的假设是将任务分成完全相等的三个部分，这样就可以实现完美平衡的并行处理，但这需要每个项目具有相同的恒定工作量，并且需要一个完美的调度程序在整个操作期间分配给每个线程相同的CPU时间，而且永远不会被其他线程/应用程序干扰。在实践中，这种情况从未发生过。因此，创建n个工作线程并在工作线程没有任务时进行分割的F/J策略更适合实际情况，并同时处理不平衡的分割。 - Holger

是的，我说错了，我指的是ForEachTask和AbstractTask，它们在Streams实现中。它们使用的分割策略基于拆分目标大小。目标大小设置为完整大小除以并行级别的四倍。对于三个工作线程的拆分（1/4、1/4、1/2），比（1/3、1/3、1/3）不太优化---这就是我的意思。 - Marko Topolnik

在同时处理第一批小块的过程中，可以收集足够的信息来确定后续更大的块是否也应该进行细分。也许在未来的版本中它们会利用这一点。我曾经在lambda-dev上询问过关于AbstractSpliterator中使用的低效策略的政策问题。但正如你所指出的，一个块的时间并不是其他块的时间的可靠预测器。 - Marko Topolnik

也许对于随机访问源和当您的目标并行性确实是质数时，使用自定义实现会更好，但这总是与一个移动的目标进行比较。我也不喜欢API强制执行O(log(n))复杂度来进行拆分，但我对自定义实现的有用性表示怀疑。开发人员可以做很多错误的事情... - Holger