并行流与串行流上的减少与收集方法[重复]

Question

我无法理解 reduce 方法与 collect 方法的工作原理。 给定
Set<String> set = new TreeSet<>(Set.of("b", "c", "d"));

为什么下面这段代码会产生 firstResult= cdbcdb 和 secondResult= bcd

// Using reduce with StringBuilder
StringBuilder firstResult = set
        .parallelStream()
        .map(s -> new StringBuilder(s))
        // Requires Identity(StringBuilder), BiFunction, BinaryOperator
        .reduce(new StringBuilder(),
                (a, b) -> a.append(b),
                (a, b) -> a.append(b));

// Using collect with StringBuilder
StringBuilder secondResult = set
        .parallelStream()
        .map(s -> new StringBuilder(s))
        // collect requires Supplier, BiConsumer, BiConsumer
        .collect(StringBuilder::new,
                (a, b) -> a.append(b),
                (a, b) -> a.append(b));

虽然下面的代码生成 firstResult=bdc 和 secondResult=''

// Using reduce with StringBuilder
StringBuilder firstResult = set
        .parallelStream()
        .map(s -> new StringBuilder(s))
        // Requires Identity(StringBuilder), BiFunction, BinaryOperator
        .reduce(new StringBuilder(),
                (a, b) -> b.append(a),
                (a, b) -> a.append(b));


// Using collect with StringBuilder
StringBuilder secondResult = set
        .parallelStream()
        .map(s -> new StringBuilder(s))
        // collect requires Supplier, BiConsumer, BiConsumer
        .collect(StringBuilder::new,
                (a, b) -> b.append(a),
                (a, b) -> a.append(b));

非常感谢您！

Answer 1

首先，修改给定程序以查看每个场景中发生的情况。

import java.util.Set;
import java.util.TreeSet;

public class ReduceVsCollect {
    public static void main(String[] args) {
        Set<String> set = new TreeSet<>(Set.of("b", "c", "d"));
        parallelReduceA_Append_B(set);
        parallelCollectA_Append_B(set);
        parallelReduceB_Append_A(set);
        parallelCollectB_Append_A(set);
    }

    private static void parallelReduceA_Append_B(Set<String> set) {
        System.out.println("-".repeat(80));
        System.out.println("Start parallelReduceA_Append_B");
        StringBuilder identity = new StringBuilder();
        System.out.println("identity hash: " + System.identityHashCode(identity));
        StringBuilder result = set
                .parallelStream()
                .map(s -> new StringBuilder(s))
                .reduce(identity,
                        (a, b) -> {
                            System.out.println(String.format("called accumulator, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            return a.append(b);
                        },
                        (a, b) -> {
                            System.out.println(String.format("called combiner, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            return a.append(b);
                        }
                );
        System.out.println("Result: " + result);
    }

    private static void parallelCollectA_Append_B(Set<String> set) {
        System.out.println("-".repeat(80));
        System.out.println("Start parallelCollectA_Append_B");
        StringBuilder result = set
                .parallelStream()
                .map(s -> new StringBuilder(s))
                .collect(StringBuilder::new,
                        (a, b) -> {
                            System.out.println(String.format("called accumulator, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            a.append(b);
                        },
                        (a, b) -> {
                            System.out.println(String.format("called combiner, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            a.append(b);
                        });
        System.out.println("Result: " + result);
    }

    private static void parallelReduceB_Append_A(Set<String> set) {
        System.out.println("-".repeat(80));
        System.out.println("Start parallelReduceB_Append_A");
        StringBuilder identity = new StringBuilder();
        System.out.println("identity hash: " + System.identityHashCode(identity));
        StringBuilder result = set
                .parallelStream()
                .map(s -> new StringBuilder(s))
                .reduce(identity,
                        (a, b) -> {
                            System.out.println(String.format("called accumulator, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            return b.append(a);
                        },
                        (a, b) -> {
                            System.out.println(String.format("called combiner, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            return b.append(a);
                        });
        System.out.println("Result: " + result);
    }

    private static void parallelCollectB_Append_A(Set<String> set) {
        System.out.println("-".repeat(80));
        System.out.println("Start parallelCollectB_Append_A");
        StringBuilder result = set
                .parallelStream()
                .map(s -> new StringBuilder(s))
                .collect(StringBuilder::new,
                        (a, b) -> {
                            System.out.println(String.format("called accumulator, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            b.append(a);
                        },
                        (a, b) -> {
                            System.out.println(String.format("called combiner, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            b.append(a);
                        });
        System.out.println("Result: " + result);
    }
}

结果

--------------------------------------------------------------------------------
Start parallelReduceA_Append_B
identity hash: 1642360923
called accumulator, a hash:1642360923, a:, b:c
called accumulator, a hash:1642360923, a:, b:b
called accumulator, a hash:1642360923, a:cb, b:d
called combiner, a hash:1642360923, a:cbd, b:cbd
Result: cbdcbd
--------------------------------------------------------------------------------
Start parallelCollectA_Append_B
called accumulator, a hash:1940447180, a:, b:c
called accumulator, a hash:572416449, a:, b:b
called accumulator, a hash:1940447180, a:c, b:d
called combiner, a hash:572416449, a:b, b:cd
Result: bcd
--------------------------------------------------------------------------------
Start parallelReduceB_Append_A
identity hash: 245565335
called accumulator, a hash:245565335, a:, b:c
called accumulator, a hash:245565335, a:, b:b
called accumulator, a hash:476402209, a:c, b:d
called combiner, a hash:1490180672, a:b, b:dc
Result: dcb
--------------------------------------------------------------------------------
Start parallelCollectB_Append_A
called accumulator, a hash:358699161, a:, b:c
called accumulator, a hash:1802696844, a:, b:b
called accumulator, a hash:358699161, a:, b:d
called combiner, a hash:1802696844, a:, b:
Result: 

不同场景说明：

1。使用a.append(b) 减少结果为“cdbcdb”：

重复序列是由于在reduce(Refer to this answer) 中使用了可变的身份 - new StringBuilder()。 identity 的值在调用accumulator 时不断变化。所有中间结果也存储在identity中，所以调用combiner时会显示两个序列。

2。使用a.append(b) 收集结果为“bcd”：

作为Javadoc 状态，

对此流的元素执行可变归约操作。可变归约是一种归约值是可变结果容器（例如 ArrayList）的归约，通过更新结果的状态而不是替换结果来合并元素

因此这是使用StringBuilder的有效场景

3。使用b.append(a) 减少结果为“bdc”：

这次identity没有因为参数交换而被修改，中间结果存储在.map(s -> new StringBuilder(s))中的StringBuilder中，因此没有显示重复序列。

4。使用b.append(a) 收集结果为“”：

参考psuedocode of collect，

     R result = supplier.get();
     for (T element : this stream)
         accumulator.accept(result, element);
     return result;

在累加器中，我们调用了b.append(a)，这意味着结果(a)从不被修改，所以它的初始值保持为new StringBuilder()。