累加器在并行流中无法正常工作答案

【问题标题】：Accumulator not working properly in parallel stream累加器在并行流中无法正常工作
【发布时间】：2019-01-20 13:40:10
【问题描述】：

我制作了一个收集器，可以将流简化为一个地图，该地图将键作为某些客户可以购买的物品，并将客户的姓名作为值，我的实现在顺序流中正常工作但是当我尝试使用parallel 时，它根本不起作用，结果集总是包含一个客户名称。

List<Customer> customerList = this.mall.getCustomerList();

Supplier<Object> supplier = ConcurrentHashMap<String,Set<String>>::new;

BiConsumer<Object, Customer> accumulator = ((o, customer) -> customer.getWantToBuy().stream().map(Item::getName).forEach(
            item -> ((ConcurrentHashMap<String,Set<String>>)o)
                    .merge(item,new HashSet<String>(Collections.singleton(customer.getName())),
                            (s,s2) -> {
                                HashSet<String> res = new HashSet<>(s);
                                res.addAll(s2);
                                return res;
                            })
    ));

BinaryOperator<Object> combiner = (o,o2) -> {
        ConcurrentHashMap<String,Set<String>> res = new ConcurrentHashMap<>((ConcurrentHashMap<String,Set<String>>)o);
        res.putAll((ConcurrentHashMap<String,Set<String>>)o2);
        return res;
    };

Function<Object, Map<String, Set<String>>> finisher = (o) -> new HashMap<>((ConcurrentHashMap<String,Set<String>>)o);

Collector<Customer, ?, Map<String, Set<String>>> toItemAsKey =
        new CollectorImpl<>(supplier, accumulator, combiner, finisher, EnumSet.of(
            Collector.Characteristics.CONCURRENT,
            Collector.Characteristics.IDENTITY_FINISH));

Map<String, Set<String>> itemMap = customerList.stream().parallel().collect(toItemAsKey);

我的accumulator 实现或另一个Function 肯定存在问题，但我无法弄清楚！谁能建议我该怎么做？

【问题讨论】：

一般性问题：您真的有太多数据以至于并行执行在理论上也有意义吗？
在我的需求中我没有那么多数据，但我想练习制作CONCURENTCollectors
为什么你在地球上使用Object 作为所有函数的类型参数，而它应该是ConcurrentHashMap<String,Set<String>>，只是为了在每个函数中执行类型转换？此外，当您的实际完成函数不是标识函数时，不要指定IDENTITY_FINISH。此外，不要使用像CollectorImpl 这样的类；只需致电Collector.of(…) 即可获得收集器。
感谢您的建议，我会考虑的

标签： java java-8 java-stream collectors

【解决方案1】：

您的组合器未正确实现。
您覆盖所有具有相同键的条目。您想要的是向现有键添加值。

BinaryOperator<ConcurrentHashMap<String,Set<String>>> combiner = (o,o2) -> {
        ConcurrentHashMap<String,Set<String>> res = new ConcurrentHashMap<>(o);
        o2.forEach((key, set) -> set.forEach(string -> res.computeIfAbsent(key, k -> new HashSet<>())
                                                          .add(string)));
        return res;
    };

【讨论】：

这绝对是我在Map 中缺少的putAll 不是在组合器中使用的适当方法，谢谢