减少并行流,没有组合器正确执行多个线程.在这种情况下我应该何时使用合成器？

Question

我读过以下关于:

/sf/answers/1596992211/

我决定组合器仅在并行流中使用以获得正确的合并累加器结果.每个线程上有一个累加器实例.

因此,我做出了解决方案,减少没有合并器将无法正常工作.

为了检查这一点,我写了以下示例:

   Person reduce = Person.getPersons().stream()
                .parallel() 
                .reduce(new Person(), (intermediateResult, p2) -> {
                    System.out.println(Thread.currentThread().getName());
                    return new Person("default", intermediateResult.getAge() + p2.getAge());
                });
        System.out.println(reduce);

模型:

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;
   }
}

如您所见,我不提供组合器
样本输出:

ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-2
ForkJoinPool.commonPool-worker-1
ForkJoinPool.commonPool-worker-2
ForkJoinPool.commonPool-worker-1
ForkJoinPool.commonPool-worker-1
Person{name='default', age=72}

我已多次执行应用程序,并且总是看到正确的结果.

请解释如果没有提供组合器,如何减少并行流的工作.

Answer 1

在这种情况下,您的累加器也可以用作组合器.这是缩减类型与流元素类型相同时的简写.从而

myStream.reduce(identity, accumulator);

完全等同于

myStream.reduce(identity, accumulator, accumulator);

您甚至可以在OpenJDK中检查这些方法的源代码:

@Override
public final <R> R reduce(R identity, BiFunction<R, ? super P_OUT, R> accumulator, 
                          BinaryOperator<R> combiner) {
    return evaluate(ReduceOps.makeRef(identity, accumulator, combiner));
}

@Override
public final P_OUT reduce(final P_OUT identity, final BinaryOperator<P_OUT> accumulator) {
    return evaluate(ReduceOps.makeRef(identity, accumulator, accumulator));
}

三参数版本更灵活,因为缩减操作可能产生另一种类型的对象.在这种情况下,您不能使用双参数缩减,因为您没有提供如何组合结果类型的两个元素的规则.但是,当结果类型相同时,累加器和组合器在同一对象类型上工作,因此如果它是关联的,它应该是相同的操作.