在Stream reduce方法中，求和必须始终为0，乘法必须始终为1吗？

Question

我继续学习 java 8。

我发现了一个有趣的行为：

让我们看看代码示例：

// identity value and accumulator and combiner
Integer summaryAge = Person.getPersons().stream()
        //.parallel()  //will return surprising result
        .reduce(1,
                (intermediateResult, p) -> intermediateResult + p.age,
                (ir1, ir2) -> ir1 + ir2);
System.out.println(summaryAge);

和模型类：

public class Person {

    String name;

    Integer age;
    ///...

    public static Collection<Person> getPersons() {
        List<Person> persons = new ArrayList<>();
        persons.add(new Person("Vasya", 12));
        persons.add(new Person("Petya", 32));
        persons.add(new Person("Serj", 10));
        persons.add(new Person("Onotole", 18));
        return persons;
   }
}

12+32+10+18 = 72。对于顺序流，此代码始终返回73，即72 + 1，但对于并行，它始终返回76，即72 + 4*1（4 等于流元素计数）。

当我看到这个结果时，我认为并行流和顺序流返回不同的结果很奇怪。

我是不是在什么地方违约了？

附言

对我来说，73 是预期结果，但 76 不是。

Answer 1

@holger answer 很好地解释了不同功能的标识是什么，但没有解释为什么我们需要标识以及为什么您在 parallel 和 之间有不同的结果>顺序流。

您的问题可以简化为知道如何对 2 个元素求和的元素列表求和。

那么让我们获取一个列表L = {12,32,10,18} 和一个求和函数(a,b) -> a + b

就像你在学校学习一样，你会这样做：

(12,32) -> 12 + 32 -> 44
(44,10) -> 44 + 10 -> 54
(54,18) -> 54 + 18 -> 72

现在想象我们的列表变成L = {12}，如何对这个列表求和？身份 (x op identity = x) 来了。

(0,12) -> 12

所以现在你可以理解为什么如果你用1 而不是0 得到+1，那是因为你使用了错误的值进行初始化。

(1,12) -> 1 + 12 -> 13
(13,32) -> 13 + 32 -> 45
(45,10) -> 45 + 10 -> 55
(55,18) -> 55 + 18 -> 73

那么现在，我们如何才能提高速度？并行化事物

如果我们可以拆分我们的列表并将这些拆分后的列表提供给 4 个不同的线程（假设是 4 核 cpu）然后合并它会怎样？这将给我们L1 = {12}、L2 = {32}、L3 = {10}、L4 = {18}

所以身份 = 1

• 线程1：(1,12) -> 1+12 -> 13
• 线程2：(1,32) -> 1+32 -> 33
• 线程3：(1,10) -> 1+10 -> 11
• 线程4：(1,18) -> 1+18 -> 19

然后组合13 + 33 + 11 +19，等于76，这就解释了为什么错误会传播4次。

在这种情况下，并行可能效率较低。

但是这个结果取决于你的机器和输入列表。 Java 不会为 1000 个元素创建 1000 个线程，并且随着输入的增长，错误的传播速度会更慢。

尝试运行这段代码求和一千1s，结果非常接近 1000

public class StreamReduce {

public static void main(String[] args) {
        int sum = IntStream.range(0, 1000).map(i -> 1).parallel().reduce(1, (r, e) -> r + e);
        System.out.println("sum: " + sum);
    }
}

现在您应该明白，如果您违反身份契约，为什么会在并行和顺序之间产生不同的结果。

请参阅Oracle doc 了解正确的写法

---

问题的本质是什么？

Answer 2

我对这里的看法略有不同。尽管@user43968's answer 给出了一个合理的理由，为什么并行性需要身份，但这真的有必要吗？我相信不是因为二元运算符本身的关联性足以让我们并行化 reduce 工作。

给定一个表达式A op B op C op D，关联性保证它的求值等于(A op B) op (C op D)，这样我们就可以并行求子表达式(A op B)和(C op D)，然后合并结果而不改变最终结果。例如，对于加法运算，初始值 = 10，并且 L = [1, 2, 3]，我们要计算 10 + 1 + 2 + 3 = 16。我们应该可以计算 10 + 1 = 11 和 2 + 3 = 5 并行，最后做 11 + 5 = 16。

我能想到的 Java 要求初始值是一个标识的唯一原因是语言开发人员希望使实现简单并且所有并行化的子作业都是对称的。否则，他们可能不得不区分将初始值作为输入的第一个子作业与其他没有的子作业。现在，他们只需要将初始值平均分配给每个子作业，这本身也是一个“减少”。

但是，这更多的是关于实施限制，不应该向语言用户 IMO 提出。我的直觉告诉我必须存在一个简单的实现，不需要初始值是一个身份。

Answer 3

你的问题真的有两个部分。当您使用顺序获得 73 时，为什么使用并行获得 76。就Reduce而言，乘法和加法的身份是什么。

回答后者将有助于回答第一部分。恒等式是一个数学概念，对于那些非数学极客，我会尽量保持简单。标识是应用于自身返回相同值的值。

加性恒等式是 0。如果我们假设 a 是任意数，则数的恒等属性表明 a em> 加上它的身份将返回 a。 （基本上，a + 0=a）。乘法恒等式表示 b 乘以它的恒等式，即 1) 总是返回自身，b。

java reduce 方法使用的标识更加多变。让我们有能力说，如果我们愿意的话，我们希望通过一个额外的步骤来执行加法和乘法运算。如果你以你的例子为例：并将身份更改为 0，你将得到 72。

    Integer summaryAge = Person.getPersons().stream()
            .reduce(0, (intermediateResult, p) -> intermediateResult + p.age,
                    (ir1, ir2) -> ir1 + ir2);
    System.out.println(summaryAge);

这只是将年龄相加并返回该值。将其更改为 100，您将返回 172。但是当您并行运行时，为什么您的结果会得到 76，而在我的示例中会返回 472？这是因为当您使用流时，结果被视为一个集合，而不是单个元素。根据流上的 JavaDocs：

流通过将计算重构为聚合操作的管道，而不是对每个单独元素的命令式操作，从而促进并行执行。

为什么对集合的处理很重要，通过使用标准流（非并行或并行流），您在示例中所做的就是求和并将其处理为单个数字。因此你得到 73，将身份更改为 100，我会得到 172。但是为什么使用并行，你得到 76？或者在我的例子中是 472？因为 java 现在将集合拆分为更小的（单个）元素，将其标识（您将其表示为 1）相加，然后将结果与执行相同操作的其余元素相加。

如果您的意图是在结果中加 1，那么遵循 Tagir 的建议并在流返回后在末尾加 1 会更安全。

Answer 4

除了之前应该提到的出色答案之外，如果您想从零以外的值开始求和，您可以将初始加数移出流操作：

Integer summaryAge = Person.getPersons().stream()
        //.parallel()  //will return no surprising result
        .reduce(0, (intermediateResult, p) -> intermediateResult + p.age,
                    (ir1, ir2) -> ir1 + ir2)+1;

其他归约操作也是如此。例如，如果要计算以2 开头的产品，而不是做错.reduce(2, (a, b) -> a*b)，则可以做.reduce(1, (a, b) -> a*b)*2。只需找到您操作的真实身份，将“虚假身份”移到外面，您将获得顺序和并行情况的正确结果。

最后请注意，有更有效的方法可以解决您的问题：

Integer summaryAge = Person.getPersons().stream()
        //.parallel()  //will return no surprising result
        .collect(Collectors.summingInt(p -> p.age))+1;

或者

Integer summaryAge = Person.getPersons().stream()
        //.parallel()  //will return no surprising result
        .mapToInt(p -> p.age).sum()+1;

这里的求和是在每个中间步骤不装箱的情况下进行的，因此它可以更快。

Answer 5

标识值是一个值，例如x op identity = x。这不是 Java Streams 独有的概念，例如参见 on Wikipedia。

它列出了一些标识元素的例子，其中一些可以直接用Java代码表示，例如

• reduce("", String::concat)
• reduce(true, (a,b) -> a&&b)
• reduce(false, (a,b) -> a||b)
• reduce(Collections.emptySet(), (a,b)->{ Set<X> s=new HashSet<>(a); s.addAll(b); return s; })
• reduce(Double.POSITIVE_INFINITY, Math::min)
• reduce(Double.NEGATIVE_INFINITY, Math::max)

应该清楚的是，任意x 的表达式x + y == x 只能在y==0 时满足，因此0 是加法的标识元素。同样，1 是乘法的标识元素。

更复杂的例子是

• 减少谓词流

  reduce(x->true, Predicate::and)
  reduce(x->false, Predicate::or)
  

• 减少函数流

  reduce(Function.identity(), Function::andThen)
  

Answer 6

是的，你违反了组合函数的约定。身份是reduce 的第一个元素，必须满足combiner(identity, u) == u。引用Stream.reduce的Javadoc：

标识值必须是组合器函数的标识。这意味着对于所有u，combiner(identity, u) 等于u。

但是，您的组合器函数执行加法，1 不是加法的标识元素； 0 是。

• 将使用的身份更改为0，您不会感到惊讶：两个选项的结果将是 72。

• 为了您自己的娱乐，更改您的组合器函数以执行乘法（将恒等式保持为 1），您还会注意到两个选项的结果相同。

让我们构建一个标识既不是 0 也不是 1 的示例。给定您自己的域类，请考虑：

System.out.println(Person.getPersons().stream()
                    .reduce("", 
                            (acc, p) -> acc.length() > p.name.length() ? acc : p.name,
                            (n1, n2) -> n1.length() > n2.length() ? n1 : n2));

这会将 Person 流减少为最长的人名。

Answer 7

Stream.reduce 的 JavaDoc 文档明确指出

标识值必须是组合函数的标识

1 不是加法运算符的标识值，这就是您得到意外结果的原因。如果您使用 0（是加法运算符的标识值），那么您将从串行和并行流中得到相同的结果。