java - 将多个并行对象连接到单个列表中

Question

我有一个键值映射并迭代键，并调用服务并根据响应，我将所有响应添加到一些uberList

如何同时执行不同的操作？将 stream() 更改为 parallelStream() 可以解决问题吗？添加到 uberList 时是否同步？

这个想法是为了最小化响应时间。

List<MyClass> uberList = new LinkedList<>();

Map<String, List<MyOtherClass>> map = new HashMap();

//Populate map

map.entrySet().stream().filter(s -> s.getValue().size() > 0 && s.getValue().values().size() > 0).forEach(
   y  -> {

        // Do stuff
        if(noError) {
            uberList.add(MyClass3);
        }

   }

}  

//Do stuff on uberList

Answer 1

How can I execute the different operations concurrently?

一个线程一次只能执行一项任务。如果要同时执行多个操作，则必须将工作转移到其他线程。您可以创建新线程或使用 ExecutorService 来管理线程池、对任务进行排队并为您执行任务。

Will changing stream() to parallelStream() do the trick?

是的，确实如此。在内部，parallelStream() 使用 ForkJoinPool.commonPool() 为您运行任务。但请记住，parallelStream() 无法保证返回的流是否是并行的(但目前，当前的实现返回并行的流)

Does it synchronize when it adds to uberList?

由您决定在 forEach 管道中执行同步部分。通常，您不想在 forEach 内调用 collection.add() 来创建集合。相反，您应该使用 .map().collect(toX()) 方法。它使您从同步部分中解放出来:

1. 它不需要知道你的局部变量(在本例中是uberlist)。并且它不会在执行时修改它，有助于减少很多并发引起的奇怪错误
2. 您可以在.collect()部分自由更改集合类型。它使您可以更好地控制结果类型。
3. 与并行流一起使用时，它不需要给定集合的线程安全或同步。因为“可以实例化、填充和合并多个中间结果，以保持可变数据结构的隔离”( Read more about this here )

因此，您想要的是同时执行多个类似的服务调用并将结果收集到列表中。

您可以简单地通过并行流来完成:

uberList = map.entrySet().stream()
                         .parallel()  // Use .stream().parallel() to force parallism. The .parallelStream() does not guarantee that the returned stream is parallel stream
                         .filter(yourCondition)
                         .map(e -> yourService.methodCall(e))
                         .collect(Collectors.toList());

很酷，不是吗？

但正如我所说，默认并行流使用 ForkJoinPool.commonPool() 进行线程排队和执行。

不好的部分是如果你的 yourService.methodCall(e) 执行大量 IO 操作(例如 HTTP 调用，甚至 db 调用...)或 长正在运行的任务那么它可能会耗尽池，其他传入的任务将永远排队等待执行。

因此，通常所有其他任务都依赖于这个公共(public)池(不仅是您自己的 yourService.methodCall(e)，而且所有其他并行流)都会因排队时间而变慢。

要解决此问题，您可以在自己的 fork-join 池上强制执行并行性:

ForkJoinPool forkJoinPool = new ForkJoinPool(4); // Typically set it to Runtime.availableProcessors()
uberlist = forkJoinPool.submit(() -> {
     return map.entrySet().stream()
                             .parallel()  // Use .stream().parallel() to force parallism. The .parallelStream() does not guarantee that the returned stream is parallel stream
                             .filter(yourCondition)
                             .map(e -> yourService.methodCall(e))
                             .collect(Collectors.toList());
}).get();