使用 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）如果我启用所有调试系统输出，我可以看到在收集和减少之间执行的操作完全相同。在这两种情况下，组合器都没有被使用。
3）如果我让流并行，收集总是返回正确的结果。 reduce() 每次都给我不同的结果。
4) 我不应该在并行流中使用reduce吗？

谢谢，
保罗

Answer 1

reduce 和collect 的区别在于collect 是一种增强的归约形式，可以并行处理可变对象。 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 方法可以为您执行这些计算。但是，我认为问题更多是关于收集和归约的机制，而不是关于如何最简洁地进行平均。