reduce操作中并行流的同步问题

Question

我正在尝试使用并行流连接字符串。

StringBuffer concat = Arrays.stream(grades)
        .parallel()
        .reduce(
                new StringBuffer(),
                (sb, s) -> sb.append(s),
                (sb1, sb2) -> sb1.append(sb2)
        );

我在下面的代码中也发现了同样的问题。

List<Integer> ages = people
            .stream()
            .parallel()
            .reduce(
                Collections.synchronizedList(new ArrayList<>()),
                (list, p) -> { list.add(p.getAge()); return list; },
                (list1, list2) -> { list1.addAll(list2) ; return list1; }
            );

在这里，我还使用了一个同步集合，所有的方法都是线程安全的。

我在Java文档中看到了这个

int sum = numbers.stream().reduce(0, (x,y) -> x+y);   or:

int sum = numbers.stream().reduce(0, Integer::sum);   These reduction operations can run safely in parallel with almost no

int sum = numbers.parallelStream().reduce(0, Integer::sum);   Reduction parallellizes well because the implementation can operate on

我是不是漏掉了什么？使用线程安全的数据结构还不够吗？

Answer 1

当您执行新建StringBuffer()时，您正在创建对单个缓冲区的引用。当您执行.parallel()时，两个并行流都将传递这个引用，从而在同一个可变缓冲区上操作。空缓冲区首先用“B”缩小，然后用“A”缩小，然后缩小到自身，产生“baba”。

对于使用可变结构（如StringBuffers）进行类似的操作，可以尝试.collect():

StringBuffer concat = Arrays.stream(grades).parallel().collect(
    () -> new StringBuffer(),
    (sb, s) -> sb.append(s),
    (sb1, sb2) -> sb1.append(sb2));