当只有一个列表作为输入时，parallelStream（）创建了多少个线程？

Question

我正在处理一批100个对象，将它们分成10个分区，每个分区都被发送到一个单独的线程进行并行处理。这是当前的代码：

1. var itemsToSave = new ConcurrentLinkedQueue<ItemToSave>();
2
3. Lists.partition(originalList, 10)
4.  .parallelStream()
5.  .forEach(partitionedList -> process(partitionedList, itemsToSave));

我的理解是

line1: creates a thread safe list to add the individual items once they are processed

line3: will return a number of lists, each of them with 10 entries from originalList

line4 will spawn a new thread for each of the lists created on line3

line5: for each list in its own thread, start the process that is supposed to be parallel.

如果我的理解有误，请纠正我，但这段代码仍按预期工作：并行化前75秒处理，并行化后20秒。看起来不错。

但过了一段时间，我注意到，根据Kibana的说法，这个过程实际上需要1ms或甚至零。这是因为新代码处理项目的速度非常快，以至于数据库没有时间处理100个项目，因此应该是100个的批次现在少于10个。

从这个意义上说，前面显示的代码的第3行将返回一个单个列表，毕竟从少于10个项目的列表中包含10个项目的分区将是一个单个列表。

然后，这个列表被发送到parallelStream（）。这里我的问题是：parallelStream（）是否仍然生成一个新线程来处理一个列表？或者它只在输入多个列表时生成线程吗？因为（对我来说）打开一个新线程来处理一批项目没有多大意义……这可能会按顺序发生，并减少生成线程等的开销。

所以：当只给一个列表作为输入时，并行流（）创建了多少线程？

很抱歉提了这么长的问题，但我觉得必须解释一下我的想法

Answer 1

我想你用的是Lists.partition是番石榴的一部分

顺便说一下，parallelstream使用了ForkJoinPool。commonPool（）以获取处理所需的线程。

下面的代码行只是告诉您有多少线程可供您处理。

System.out.println(Runtime.getRuntime().availableProcessors()); 

现在，让我们举一个与您的案例类似的例子：

    List<List<String>> list =new ArrayList<List<String>>();
    List<String> w1=new ArrayList<String>();

    w1.add("A");w1.add("B");w1.add("C");w1.add("D");w1.add("E");w1.add("F");w1.add("G");w1.add("H");w1.add("I");w1.add("J"); list.add(w1);

    list.parallelStream().
    forEach(x-> System.out.println(Thread.currentThread().getName())); // this gives output as `main`, no other threads got used.

但是现在，让我们假设您为要处理的并行流提供了项数：

IntStream.range(0, 10).boxed().parallel().forEach(x-> System.out.println(Thread.currentThread().getName()));

上述输出类似于：

main
main
main
main
ForkJoinPool.commonPool-worker-2
main
main
ForkJoinPool.commonPool-worker-1
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-2

因此，决定线程数实际上取决于并行流的输入

Answer 2

当只有一个列表作为输入时，创建了多少线程？

在并行流的框架下，使用了Fork-Join框架。它将任务（输入到流中的全部元素）分成更小的子任务（可以进一步拆分），然后按照创建顺序连接已执行任务的结果（这保证了在并行处理过程中，有序流中元素的相遇顺序将保持不变，您可以在减少、收集等行为中观察到）。

在执行并行流时，来自所谓的调用公共池（延迟初始化，其最大大小取决于CPU内核的实际数量）的线程将被分配任务。

占用多少线程取决于输入的大小、拆分器的实现方式（即任务如何划分为子任务）、公共池中有多少线程（取决于实际机器）。同样重要的是，处理每个项目的速度有多快（即取决于您的逻辑），线程在完成前一个任务时会尽量通过窃取工作来做到最好。

Javadoc说“返回一个可能并行的流”。甚至不能保证会有多个线程占用您的任务（如果元素数量很小，那么一个线程可能会更快地处理它，而不需要分割和连接结果）。

您不应该关心线程的实际数量，因为您无法控制是否。唯一的问题是，您需要决定并行流是否可以并行执行流。