使用并行流时,减少行为很奇怪,但在Java 8u5中适用于顺序流

Question

class Foo{
    int len;
}
public class Main {
    public static void main(String[] args) throws Exception{
    System.out.println(Stream.of("alpha", "beta", "gamma", "delta").parallel().reduce(
            new Foo(),
            (f, s) -> { f.len += s.length(); return f; },
            (f1, f2) -> {
                Foo f = new Foo();
                /* check self-reduction
                if (f1 == f2) { 
                    System.out.println("equal");
                    f.len = f1.len;
                    return f;
                }
                */
                f.len = f1.len + f2.len;
                return f;
            }
    ).len);
}

代码尝试计算几个字符串的总长度.

这段代码仅在
1.I使用顺序流(通过删除"parallel()"函数调用)
或
2.I使用Integer而不是Foo 时才打印19 ,这只是一个int的包装器.

否则控制台将打印20或36.为了调试这个问题,我添加了代码"check self-reduction",它确实改变了输出:"equal"总是被打印两次.控制台有时会打印8,有时打印10.

我的理解是reduce()是并行foldr/foldl的Java实现.reduce()的第三个参数,combiner用于合并缩减的并行执行结果.是对的吗？如果是这样,为什么减少的结果需要与自身结合？此外,如何修复此代码,以便它提供正确的输出并仍然并行运行？

编辑:请忽略我没有使用方法参考来简化代码的事实,因为我的最终目标是通过向Foo添加更多字段来压缩.

Answer 1

你的代码非常糟糕.您正在使用减速器功能,该功能未能满足累加器/组合器功能是关联,无状态和无干扰的要求.而且可变的Foo不是减少的身份.当并行执行时,所有这些都可能导致错误的结果.

你也要比你需要的更难!试试这个:

int totalLen = 
    Stream.of(... stuff ...)
          .parallel()
          .mapToInt(String::length)
          .sum();

要么

int totalLen = 
    Stream.of(... stuff ...)
          .parallel()
          .mapToInt(String::length)
          .reduce(0, Integer::sum);

此外,您正在尝试使用reduce减少值(这就是它的工作原理Integer),但您尝试使用可变状态容器来减少结果.如果要将其缩减为可变状态容器(如a List或StringBuilder),请collect()改为使用,这是为突变而设计的.