java - 在并行流上调用顺序使所有先前的操作顺序

Question

我有一组重要的数据，我想调用缓慢但干净的方法，而不是调用对第一个方法的结果有副作用的快速方法。我对中间结果不感兴趣，所以我不想收集它们。

明显的解决方案是创建并行流，进行慢速调用，再次使流顺序化，然后进行快速调用。问题是，所有代码都在单线程中执行，没有实际的并行性。

示例代码:

@Test
public void testParallelStream() throws ExecutionException, InterruptedException
{
    ForkJoinPool forkJoinPool = new ForkJoinPool(Runtime.getRuntime().availableProcessors() * 2);
    Set<String> threads = forkJoinPool.submit(()-> new Random().ints(100).boxed()
            .parallel()
            .map(this::slowOperation)
            .sequential()
            .map(Function.identity())//some fast operation, but must be in single thread
            .collect(Collectors.toSet())
    ).get();
    System.out.println(threads);
    Assert.assertEquals(Runtime.getRuntime().availableProcessors() * 2, threads.size());
}

private String slowOperation(int value)
{
    try
    {
        Thread.sleep(100);
    }
    catch (InterruptedException e)
    {
        e.printStackTrace();
    }
    return Thread.currentThread().getName();
}

如果我删除 sequential，代码将按预期执行，但显然，非并行操作将在多个线程中调用。

您能否推荐一些关于此类行为的引用资料，或者一些避免临时集合的方法？

Answer 1

将流从 parallel() 切换到 sequential() 在最初的 Stream API 设计中可行，但引起了很多问题，最终实现是 changed ，所以它只是为整个管道打开和关闭并行标志。目前的文档确实含糊不清，但在Java-9中得到了改进:

The stream pipeline is executed sequentially or in parallel depending on the mode of the stream on which the terminal operation is invoked. The sequential or parallel mode of a stream can be determined with the BaseStream.isParallel() method, and the stream's mode can be modified with the BaseStream.sequential() and BaseStream.parallel() operations. The most recent sequential or parallel mode setting applies to the execution of the entire stream pipeline.

至于您的问题，您可以将所有内容收集到中间 List 中并开始新的顺序管道:

new Random().ints(100).boxed()
        .parallel()
        .map(this::slowOperation)
        .collect(Collectors.toList())
        // Start new stream here
        .stream()
        .map(Function.identity())//some fast operation, but must be in single thread
        .collect(Collectors.toSet());