如何使用 java 并行流而不是 executorThreadsPool？

Question

我想编写一个对我的 API 执行许多并行调用的测试。

ExecutorService executor = Executors.newCachedThreadPool();
        final int numOfUsers = 10;
        for (int i = 0; i < numOfUsers; i++) {
            executor.execute(() -> {
                final Device device1 = getFirstDevice();
                final ResponseDto responseDto = devicesServiceLocal.acquireDevice(device1.uuid, 4738);
                if (responseDto.status == Status.SUCCESS)
                {
                    successCount.incrementAndGet();
                }
            });
        }

我知道我可以使用 executorThreadsPool 做到这一点，如下所示：

devicesList.parallelStream()
           .map(device -> do something)

我可以用 java8 并行流创建它：

我怎样才能在一台设备上做到这一点？

意思是我想要几个电话来获取相同的设备。

类似这样的：

{{device}}.parallelStream().execute(myAction).times(10)

Answer 1

可以，但是……

你会想

Stream.generate(() -> device)
      .limit(10)
      .parallel()
      .forEach(device -> device.execute());

应该做的工作。但是NO，因为原因（我真的不知道为什么，没有头绪）。 如果我让device.execute() 等一下，然后让它打印一些东西。流每秒打印 10 次。所以它根本不是平行的，不是你想要的。

Google 是我的朋友，我发现很多文章都警告不要使用并行流。但我的目光落在了http://blog.jooq.org/2014/06/13/java-8-friday-10-subtle-mistakes-when-using-the-streams-api/ 8 号和 9 号上。8 说如果它有一个集合支持，你必须对其进行排序，它会神奇地工作：

Stream.generate(() -> device)
      .limit(10)
      .sorted((a,b)->0) // Sort it (kind of), what??
      .parallel()
      .forEach(device -> device.execute());

现在它在一秒钟后打印 8 次，在另一秒钟后打印 2 次。我有 8 个内核，所以这是我们（有点）期望的。

我在我的信息流中使用了.forEach()，但起初我（就像你的例子一样）使用.map()。 .map() 没有打印任何东西：流从未被消耗（参见链接文章中的 9）。

因此，请注意使用流，尤其是并行流。您必须确保您的流已被消耗，它是有限的 (.limit())，它正在并行工作等。流很奇怪，我建议保留您的工作解决方案。

注意：如果device.execute() 是一个阻塞操作（IO、网络...），您将永远不会有超过您将同时执行的核心数量（在我的情况下为 8 个）任务。

更新（感谢Holger）：

Holger 给出了一个优雅的替代方案：

IntStream.range(0,10)
      .parallel()
      .mapToObject(i -> getDevice())
      .forEach(device -> device.execute());

// Or shorter:
IntStream.range(0,10)
      .parallel()
      .forEach(i -> getDevice().execute());

这就像一个并行的 for 循环（并且有效）。