Java 8 流组合器从未调用过

Question

我正在编写一个自定义的 java 8 收集器，它应该计算具有 getValue() 方法的 POJO 的平均值。代码如下：

public static Collector<BoltAggregationData, BigDecimal[], BigDecimal> avgCollector = new Collector<BoltAggregationData, BigDecimal[], BigDecimal>() {

        @Override
        public Supplier<BigDecimal[]> supplier() {
            return () -> {
                BigDecimal[] start = new BigDecimal[2];
                start[0] = BigDecimal.ZERO;
                start[1] = BigDecimal.ZERO;
                return start;
            };
        }

        @Override
        public BiConsumer<BigDecimal[], BoltAggregationData> accumulator() {
            return (a,b) ->  {
                a[0] = a[0].add(b.getValue());
                a[1] = a[1].add(BigDecimal.ONE);
            };
        }

        @Override
        public BinaryOperator<BigDecimal[]> combiner() {
            return (a,b) -> {
                a[0] = a[0].add(b[0]);
                a[1] = a[1].add(b[1]);
                return a;
            };
        }

        @Override
        public Function<BigDecimal[], BigDecimal> finisher() {
            return (a) -> {
                return a[0].divide(a[1], 6 , RoundingMode.HALF_UP);
            };
        }

        private final Set<Characteristics> CHARACTERISTICS = new HashSet<Characteristics>(Arrays.asList(Characteristics.CONCURRENT, Characteristics.UNORDERED));

        @Override
        public Set<Characteristics> characteristics() {
            return CHARACTERISTICS;
        }

    };

在非并行情况下一切正常。但是，当我使用parallelStream() 时，它有时不起作用。例如，给定从 1 到 10 的值，它会计算（53/9 而不是 55/10）。调试时，调试器永远不会命中 combiner() 函数中的断点。我需要设置某种标志吗？

Answer 1

嗯，这正是您在指定 Characteristics.CONCURRENT 时所要求的：

表示这个收集器是并发的，意味着结果容器可以支持累加器函数被多个线程与同一个结果容器并发调用。

如果不是这样，就像你的 Collector 一样，你不应该指定那个标志。

---

作为旁注，new HashSet<Characteristics>(Arrays.asList(Characteristics.CONCURRENT, Characteristics.UNORDERED)); 在指定特征方面效率很低。你可以使用EnumSet.of(Characteristics.CONCURRENT, Characteristics.UNORDERED)。当您删除错误的并发特性时，您可以使用EnumSet.of(Characteristics.UNORDERED) 或Collections.singleton(Characteristics.UNORDERED)，但HashSet 绝对是矫枉过正。

Answer 2

看起来问题出在 CONCURRENT 特征上，它的作用超出了您的想象：

表示这个收集器是并发的，意思是 结果容器可以支持累加器功能 与来自多个的相同结果容器同时调用 线程。

而不是调用组合器，累加器被并发调用，对所有线程使用相同的BigDecimal[] a。 a 的访问不是原子的，所以会出错：

Thread1 -> retrieves value of a[0]: 3
Thread2 -> retrieves value of a[0]: 3
Thread1 -> adds own value: 3 + 3 = 6
Thread2 -> adds own value: 3 + 4 = 7
Thread1 -> writes 6 to a[0]
Thread2 -> writes 7 to a[0]

将 a[0] 的值设置为 7 而应该是 10。a[1] 可能会发生同样的事情，因此结果可能会不一致。

---

如果您删除 CONCURRENT 特征，则将使用组合器。