从未调用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;
        }

    };

在非并行情况下，这一切都很好。但是，当我使用并行流（）时，它有时不起作用。例如，给定从1到10的值，它会计算（53/9而不是55/10）。调试时，调试器永远不会碰到组合器（）函数中的断点。是否需要设置某种标志？

Answer 1

好的，这正是您在指定特性时所要求的。并发：

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

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

另请注意，new HashSet

Answer 2

问题似乎是并发的特性，它做了一些你想象不到的事情：

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

不是调用组合器，而是同时调用累加器，对所有线程使用相同的BigDecimal[]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]

当值应为10时，将值设为[0]7。同样的事情也可能发生在[1]上，因此结果可能不一致。

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