并发执行:Future vs parallelstream

Question

我编写了一个可调用的函数来轮询远程客户端以获取信息,并以List形式返回该信息.我正在使用threadpoolexecutor,for循环,以及Future与多个远程客户端并行执行任务.然后我将所有Future列表与addAll()组合在一起,并使用巨型组合列表.

我的问题是,使用parallelstream()会比使用future和for循环更有效吗？编码肯定更容易!如果我走那条路,我会不再需要threadpoolexecutor？

谢谢!

        for(SiteInfo site : active_sites) {
            TAG_SCANNER scanr = new TAG_SCANNER(site, loggr);
            Future<List<TagInfo>> result = threadmaker.submit(scanr);

            //SOUND THE ALARMS
            try {
                alarm_tags.addAll(result.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }

可能的解决方案代 Netbeans在这些方面提出了一些建议

active_sites.parallelstream().map((site) -> new TAG_SCANNER(site, loggr)).map((scanr) -> threadmaker.submit(scanr)).forEach((result) -> {
            //SOUND THE ALARMS
            try {
                alarm_tags.addAll(result.get());
            }
            catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });

Answer 1

这里有几个误解.首先,如果Future.get在提交任务后立即调用,使用异步任务不会提高资源利用率,在提交下一个任务之前立即等待其完成.

其次,Netbeans进行的代码转换产生了一个大致相同的代码,仍然将任务提交给一个Executor所以它不是"Future vs parallelstream"的问题,因为你只是使用并行流执行提交(和等待)并且仍然使用执行程序.由于你的第一个错误,并行执行可能会提高吞吐量,但除此之外,将两个错误结合起来让它们自行取消并不是一个好主意,它仍然是一个糟糕的解决方案:

Stream API的标准实现针对CPU绑定任务进行了优化,创建了许多与CPU核心数匹配的线程,并且在这些线程在等待操作中被阻塞时不生成新线程.因此,使用并行流执行I/O操作,或者通常可以等待的操作,不是一个好的选择.而且您无法控制实现使用的线程.

更好的选择是ExecutorService根据您对远程客户端的预期I/O带宽来配置.但是你应该在提交后立即修复等待的错误,首先提交所有任务,然后等待所有任务完成.请注意,您可以使用流API,而不是为了更好的并行性,但可能会提高可读性:

// first, submit all tasks, assuming "threadmaker" is an ExecutorService
List<Future<List<TagInfo>>> futures=threadmaker.invokeAll(
    active_sites.stream()
        .map(site -> new TAG_SCANNER(site, loggr))
        .collect(Collectors.toList())
);
// now fetch all results
for(Future<List<TagInfo>> result: futures) {
    //SOUND THE ALARMS
    try {
        alarm_tags.addAll(result.get());
    } catch (InterruptedException | ExecutionException e) {
        // not a recommended way of handling
        // but I keep your code here for simplicity
        e.printStackTrace();
    }
}

请注意,此处使用的流API是顺序的,仅用于将列表转换SiteInfo为列表Callable<List<TagInfo>>,但您可以使用循环执行相同操作.

Answer 2

通常parallelstream由非常聪明的程序员编写以非常有效地进行并行处理。

这样，与所有其他 Java 线程（例如并发包）一样，除非您是该主题的专家，否则如果您自己编写它，您可能会：

• 跑得更慢
• 介绍错误
• 有更复杂/更难遵循/等代码

换句话说：是的，使用 parallelstream。