java - 如何使用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());

应该可以完成这项工作。但是不，因为原因(我真的不知道为什么，没有头绪)。如果我让 device.execute() 等待一秒钟，然后让它打印一些东西。该流每秒打印 10 次内容。所以它根本不是并行的，不是你想要的。

Google 是我的 friend ，我发现了很多警告并行流的文章。但我的目光落在了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 循环(并且它有效)。