Java 8：并行 FOR 循环答案

【问题标题】：Java 8: Parallel FOR loopJava 8：并行 FOR 循环
【发布时间】：2015-02-17 22:48:13
【问题描述】：

我听说 Java 8 提供了很多关于并发计算的实用程序。因此，我想知道并行化给定 for 循环的最简单方法是什么？

public static void main(String[] args)
{
    Set<Server> servers = getServers();
    Map<String, String> serverData = new ConcurrentHashMap<>();

    for (Server server : servers)
    {
        String serverId = server.getIdentifier(); 
        String data = server.fetchData();

        serverData.put(serverId, data);
    }
}

【问题讨论】：

标签： java for-loop concurrency java.util.concurrent concurrent-programming

【解决方案1】：

复制'n'paste的简单示例（上面的示例使用类Server，这是由OP编写的自定义类）：

import java.io.Console;
import java.util.ArrayList;

ArrayList<String> list = new ArrayList<>();
list.add("Item1");
list.add("Item2");
list.parallelStream().forEach((o) -> {
    System.out.print(o);
});

控制台输出。随着一切并行执行，顺序可能会有所不同：

Item1
Item2

.parallelStream() 方法是在Java v8 中引入的。此示例使用JDK v1.8.0_181 进行了测试。

【讨论】：

【解决方案2】：

使用我的 Parallel.For，您的代码可能如下所示，

public staic void main(String[] args)
{
    Set<Server> servers = getServers();
    Map<String, String> serverData = new ConcurrentHashMap<>();

    Parallel.ForEach(servers, new LoopBody<Server>()
    {
        public void run(Server server)
        {
             String serverId = server.getIdentifier(); 
             String data = server.fetchData();

             serverData.put(serverId, data);
        }
    });
}

【讨论】：

那是 C#，不是 Java

【解决方案3】：

阅读streams，他们都是新的风靡一时。

特别注意并行性：

“具有显式 for 循环的处理元素本质上是串行的。流通过将计算重新构建为聚合操作的管道而不是对每个单独元素的命令式操作来促进并行执行。所有流操作都可以串行或串行执行并行。”

回顾一下，没有并行的 for 循环，它们本质上是串行的。然而，流可以完成这项工作。看看下面的代码：

    Set<Server> servers = getServers();
    Map<String, String> serverData = new ConcurrentHashMap<>();

    servers.parallelStream().forEach((server) -> {
        serverData.put(server.getIdentifier(), server.fetchData());
    });

【讨论】：

请注意，并行流有一个开销：它并不总是比串行流提高性能（或每个都正常）。
如何使用 .parallelStream() 处理线程？它是否使用任何内部线程池？
如果我们在服务器中有一百万个元素怎么办？这会创建一百万个并行流吗？
@SonuMishra，否。并行流尝试利用处理器上的多个内核。
我认为您可以更清楚地说明 forEach 方法的回调可以/将被并行调用

【解决方案4】：

更优雅或功能性的解决方案将只使用 Collectors toMap 或 toConcurrentMap 函数，避免为 ConcurrentHashMap 维护另一个有状态变量，如下例所示：

final Set<Server> servers = getServers();
Map<String, String> serverData = servers.parallelStream().collect(
    toConcurrentMap(Server::getIdentifier, Server::fetchData));

注意： 1.那些功能接口（@987654323@）不允许在这里抛出检查异常， 2. 为了充分发挥并行流的优势，服务器的数量会很大，不涉及I/O，这些功能中纯粹是数据处理(getIdentifier, fetchData)

请参考 Collectors javadoc http://docs.oracle.com/javase/8/docs/api/java/util/stream/Collectors.html#toConcurrentMap

【讨论】：

【解决方案5】：

那就是使用Stream:

servers.parallelStream().forEach(server -> {
    serverData.put(server.getIdentifier(), server.fetchData());
});

我怀疑Collector 在这里可以起到更大的作用，因为您使用并发集合。

【讨论】：

好吧，根据您的建议，我现在可以阅读 lambda 表达式以及了解流的一些细节；)