如何在并行流上使用 java map-reduce 组合器答案

【问题标题】：how to use java map-reduce combiner on parallel stream如何在并行流上使用 java map-reduce 组合器
【发布时间】：2016-02-03 14:30:28
【问题描述】：

我有这个java代码获取字符串（请求）获取 http 响应 --> 将其传递给与保存在内存中的旧响应进行比较。

我想收集 n 个比较结果到 Result 对象中

最终我想将 m 个报表对象聚合为一个对象。

我有这个代码

但我不确定我是否在执行第二个功能

我应该每次都创建新报告吗？

该机制如何运作？

        Report report = requestsList
                .parallelStream()
                .map(request ->
                                getResponse(request, e2EResultLongBL, e2EResultLongFresh)
                )
                .map(response -> compareToBl(response, e2EResultLongBL))
                .reduce(null,
                        (sumReport, compare2) ->
                        {
                            if (sumReport == null)
                            {
                                sumReport = new Report();
                            }
                            sumReport.add(compare2);
                            return  sumReport;
                        },
                        (report1, report2) ->
                        {
                            Report report3 = new Report();
                            report3.add(report2);
                            return report3;
                        });

编辑

我尝试过“收集”而不是“减少”

我收到了这个错误：

Error:(174, 21) java: no suitable method found for collect(<nulltype>,(sumReport[...]rt; },(report1,r[...]t3; })
    method java.util.stream.Stream.<R>collect(java.util.function.Supplier<R>,java.util.function.BiConsumer<R,? super com.waze.routing.automation.dataModel.ComparisonResult>,java.util.function.BiConsumer<R,R>) is not applicable
      (cannot infer type-variable(s) R
        (argument mismatch; unexpected return value))
    method java.util.stream.Stream.<R,A>collect(java.util.stream.Collector<? super com.waze.routing.automation.dataModel.ComparisonResult,A,R>) is not applicable
      (cannot infer type-variable(s) R,A
        (actual and formal argument lists differ in length))

【问题讨论】：

不要这样使用reduce。了解collect...
你能解释一下为什么不吗？
reduce 适用于类似值的类，并且不得修改其输入对象（无论如何它最适用于不可变对象）。 collect 正是为了解决您遇到的问题：您不想不必要地创建新的 Report 实例，但仅在需要时。你想做的是“Mutable Reduction”。据我们可以从您的代码示例中猜到，类似 .collect(Report::new, Report::add, Report::add); 的东西代替您的 reduce 操作应该可以工作......
我在哪里可以看到reduce operation - must not modify its input objects 的原因？
点击我已经给你的链接。向上滚动以查看“减少”部分（并与“可变减少”进行比较）。

标签： java dictionary concurrency mapreduce java-stream

【解决方案1】：

你肯定想要一个收藏家。您将为每个并行执行创建一个报告，将单个比较添加到报告中，然后将这些报告与最终函数合并。

无需在合并器上创建新报告。只需将 2 加到 1。

.collect(Collector.of(
    () -> new Report(),
    (report, compare2) -> {
       report.add(compare2);
       return report;
    },
    (report1, report2) -> {
       report1.add(report2);
       return report1;
    });

另外，如果您想让代码更简洁，请使用添加函数 return this; 您将能够将 lambda 替换为函数引用。

.collect(Collector.of(Report::new, Report::add, Report::add));

【讨论】：