java - 使用 reduce 和 collect 求平均值

Question

我正在尝试了解新的 Java 8 Stream API。

http://docs.oracle.com/javase/tutorial/collections/streams/reduction.html

我找到了使用 collect API 查找数字平均值的示例。但我觉得，同样可以使用 reduce() 来完成。

public class Test {

    public static void main(String[] args) {
        // Using collect
        System.out.println(Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
            .collect(Averager::new, Averager::accept, Averager::combine)
            .average());

        // Using reduce
        System.out.println(Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
            .reduce(new Averager(), (t, u) -> {
                t.accept(u);
                return t;
            }, (t, u) -> {
                t.combine(u);
                return t;
            }).average());
    }

    private static class Averager {
        private int total = 0;
        private int count = 0;

        public Averager() {
            // System.out.println("Creating averager");
        }

        public double average() {
            // System.out.println("Finding average");
            return count > 0 ? ((double) total) / count : 0;
        }

        public void accept(int i) {
            // System.out.println("Accepting " + i);
            total += i;
            count++;
        }

        public void combine(Averager other) {
            // System.out.println("Combining the averager : " + other);
            total += other.total;
            count += other.count;
        }

        @Override
        public String toString() {
            return "[total : " + total + ", count: " + count + "]";
        }
    }
}

1) 有什么理由我应该在这里使用 collect 而不是 reduce？
2) 如果我启用所有调试系统输出，我可以看到执行的操作在 collect 和 reduce 之间完全相同。在这两种情况下，组合器根本没有被使用。
3)如果我使流平行，收集总是返回我正确的结果。 reduce() 每次都给我不同的结果。
4) 我不应该在并行流中使用 reduce 吗？

谢谢，
保罗

Answer 1

reduce 和collect 的区别在于collect 是reduction 的一种增强形式，可以并行处理可变对象。 collect 算法线程限制各种结果对象，因此即使它们不是线程安全的，也可以安全地改变它们。这就是 Averager 使用 collect 工作的原因。对于使用 reduce 的顺序计算，这通常无关紧要，但对于并行计算，正如您观察到的那样，它会给出不正确的结果。

关键是 reduce 只要处理值 就可以工作，而不是可变对象。您可以通过查看 reduce 的第一个参数来了解这一点。示例代码传递了 new Averager()，它是一个单个对象，在并行缩减中被多个线程用作标识值。并行流的工作方式是将工作负载分成由各个线程处理的片段。如果多个线程正在改变同一个(非线程安全的)对象，应该清楚为什么这会导致不正确的结果。

可以使用reduce 来计算平均值，但您需要使累积对象不可变。考虑一个对象 ImmutableAverager:

static class ImmutableAverager {
    private final int total;
    private final int count;

    public ImmutableAverager() {
        this.total = 0;
        this.count = 0;
    }
    
    public ImmutableAverager(int total, int count) {
        this.total = total;
        this.count = count;
    }

    public double average() {
        return count > 0 ? ((double) total) / count : 0;
    }

    public ImmutableAverager accept(int i) {
        return new ImmutableAverager(total + i, count + 1);
    }

    public ImmutableAverager combine(ImmutableAverager other) {
        return new ImmutableAverager(total + other.total, count + other.count);
    }
}

请注意，我已经调整了 accept 和 combine 的签名以返回一个新的 ImmutableAverager 而不是改变 this。 (这些更改还使方法与 reduce 的函数参数相匹配，因此我们可以使用方法引用。)您可以像这样使用 ImmutableAverager:

    double average = Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
            .parallel()
            .reduce(new ImmutableAverager(), 
                    ImmutableAverager::accept,
                    ImmutableAverager::combine)
            .average();
    System.out.println("Average: "+average);

将不可变值对象与 reduce 结合使用应该可以并行给出正确的结果。

最后，请注意 IntStream 和 DoubleStream 有 summaryStatistics() 方法，Collectors 有 averagingDouble 、averagingInt 和 averagingLong 方法可以为您进行这些计算。但是，我认为问题更多的是关于收集和减少的机制，而不是关于如何最简洁地进行平均。