Java 8流 - collect与reduce

Question

Java 8流 - collect与reduce

javajava-8java-stream

175

何时使用 collect() 和 reduce()？是否有好的、具体的例子说明什么情况下肯定更好地采用其中一种方法？

Javadoc提到collect()是一种可变还原操作。

考虑到它是可变还原操作，我认为它需要同步（内部），这反过来可能会损害性能。据推测,reduce() 更容易并行化，但需要在 reduce 的每一步中创建一个新数据结构以返回结果。

上述说法纯属猜测，我希望专家们能在这里发表意见。

- jimhooker2002

1

你链接的页面剩下的部分已经解释了：与reduce()一样，以这种抽象方式表达collect的好处是它直接适用于并行化：我们可以并行累积部分结果，然后再组合它们，只要累积和组合函数满足适当的要求即可。 - JB Nizet

3

请参阅Angelika Langer的“Java 8中的流：Reduce vs. Collect” - https://www.youtube.com/watch?v=oWlWEKNM5Aw。 - MasterJoe

8个回答

57

原因很简单：

collect() 只能用于可变的结果对象。
reduce() 设计用于不可变的结果对象。

使用不可变结果对象的 "`reduce()`" 示例

public class Employee {
  private Integer salary;
  public Employee(String aSalary){
    this.salary = new Integer(aSalary);
  }
  public Integer getSalary(){
    return this.salary;
  }
}

@Test
public void testReduceWithImmutable(){
  List<Employee> list = new LinkedList<>();
  list.add(new Employee("1"));
  list.add(new Employee("2"));
  list.add(new Employee("3"));

  Integer sum = list
  .stream()
  .map(Employee::getSalary)
  .reduce(0, (Integer a, Integer b) -> Integer.sum(a, b));

  assertEquals(Integer.valueOf(6), sum);
}

"`collect()` with mutable" 示例

例如，如果您想使用 collect() 手动计算一个总和，它不能使用 BigDecimal，而只能使用来自 org.apache.commons.lang.mutable 的 MutableInt。请参见：

public class Employee {
  private MutableInt salary;
  public Employee(String aSalary){
    this.salary = new MutableInt(aSalary);
  }
  public MutableInt getSalary(){
    return this.salary;
  }
}

@Test
public void testCollectWithMutable(){
  List<Employee> list = new LinkedList<>();
  list.add(new Employee("1"));
  list.add(new Employee("2"));

  MutableInt sum = list.stream().collect(
    MutableInt::new, 
    (MutableInt container, Employee employee) -> 
      container.add(employee.getSalary().intValue())
    , 
    MutableInt::add);
  assertEquals(new MutableInt(3), sum);
}

这是因为accumulator container.add(employee.getSalary().intValue());不会返回一个新的结果对象，而是要改变类型为MutableInt的可变container的状态。

如果您想要使用BigDecimal替代container，则不能使用collect()方法，因为container.add(employee.getSalary());不会改变container，因为BigDecimal是不可变的。（除此之外，BigDecimal::new也无法使用，因为BigDecimal没有空构造函数。）

- Sandro

3

请注意，您正在使用Integer构造函数（new Integer(6)），该构造函数在后续Java版本中已被弃用。 - MC Emperor

1

好的发现 @MCEmperor！我已将其更改为 Integer.valueOf(6)。 - Sandro

@Sandro - 我有点困惑。你为什么说collect()只能用于可变对象？我用它来连接字符串了。字符串 allNames = employees.stream() .map(Employee::getNameString) .collect(Collectors.joining(", ")) .toString(); - MasterJoe

1

@MasterJoe2 这很简单。简而言之 - 实现仍然使用可变的 StringBuilder。请参见：http://hg.openjdk.java.net/jdk8/jdk8/jdk/file/687fd7c7986d/src/share/classes/java/util/stream/Collectors.java#l263 - Sandro

40

普通的规约操作旨在将两个不可变值（例如 int、double 等）组合成一个新值；这是一种不可变规约操作。相比之下，collect 方法旨在通过修改容器来累积要生成的结果。

为了说明问题，让我们假设您想使用简单的规约操作实现 Collectors.toList() ，如下所示：

List<Integer> numbers = stream.reduce(
        new ArrayList<Integer>(),
        (List<Integer> l, Integer e) -> {
            l.add(e);
            return l;
        },
        (List<Integer> l1, List<Integer> l2) -> {
            l1.addAll(l2);
            return l1;
        });

这相当于Collectors.toList()。然而，在这种情况下，您会改变List<Integer>的状态。我们知道ArrayList不是线程安全的，也不能在迭代时添加/删除值，因此当您更新列表或组合器尝试合并列表时，您将获得并发异常、ArrayIndexOutOfBoundsException或任何类型的异常（特别是在并行运行时）。因为您通过累加（添加）整数来改变列表的状态。如果要使此代码线程安全，则需要每次传递一个新的列表，这会影响性能。

相比之下，Collectors.toList()的工作方式类似。但是，它保证了在将值累积到列表时的线程安全。根据collect方法的文档：

使用收集器对此流的元素执行可变归约操作。如果流是并行的，并且收集器是并发的，并且流是无序的或收集器是无序的，那么将执行并发归约。 在并行执行时，可能会实例化、填充和合并多个中间结果，以维护可变数据结构的隔离。 因此，即使在使用非线程安全数据结构（例如ArrayList）进行并行归约时，也不需要额外的同步。

所以回答您的问题：

何时使用collect()与reduce()？

如果您有不可变的值，例如ints、doubles、Strings，那么普通的归约就可以正常工作。然而，如果您必须将值减少到List（可变数据结构），则需要使用带有collect方法的可变归约。

- george

在这段代码片段中，我认为问题在于它会获取标识（在这种情况下是 ArrayList 的单个实例）并假定它是“不可变的”，因此他们可以启动 x 个线程，每个线程都会“添加到标识”然后合并在一起。这是一个很好的例子。 - rogerdpack

为什么我们会得到并发修改异常，调用流只会返回串行流，这意味着它将由单个线程处理，并且组合器函数根本不会被调用？ - amarnath harish

public static void main(String[] args) { List<Integer> l = new ArrayList<>(); l.add(1); l.add(10); l.add(3); l.add(-3); l.add(-4); List<Integer> numbers = l.stream().reduce( new ArrayList<Integer>(), (List<Integer> l2, Integer e) -> { l2.add(e); return l2; }, (List<Integer> l1, List<Integer> l2) -> { l1.addAll(l2); return l1; });for(Integer i:numbers)System.out.println(i);} 我尝试过了，没有得到CCm异常。 - amarnath harish

1

@amarnathharish 当您尝试并行运行它并且多个线程尝试访问同一列表时，问题就会出现。 - george

14

让这个流程成为 a <- b <- c <- d。

在归约中，

你将会得到 ((a # b) # c) # d 的结果，其中 # 是你想要执行的操作。

在收集过程中，

你的收集器将具有某种类型的收集结构 K。

K 接受 a。然后 K 接受 b。接着 K 接受 c。最后 K 接受 d。

最后，你会询问 K 最终结果是什么。

K 然后将其给出。

- Yan Ng

3

在运行时，它们的潜在内存占用差异非常大。虽然 collect() 收集并将所有数据放入集合中，但 reduce() 明确要求您指定如何缩小通过流传递的数据。

例如，如果您想从文件中读取一些数据、处理它并将其放入某个数据库中，则可能会得到类似于以下 Java 流代码：

streamDataFromFile(file)
            .map(data -> processData(data))
            .map(result -> database.save(result))
            .collect(Collectors.toList());

在这种情况下，我们使用collect()强制Java通过流将数据流，并使其保存到数据库中。如果没有使用collect()，则数据将不会被读取也不会被存储。

如果文件大小足够大或堆大小足够低，则此代码将快乐地生成java.lang.OutOfMemoryError: Java heap space运行时错误。显而易见的原因是它尝试将通过流传递的所有数据（实际上已经存储在数据库中）堆叠到生成的集合中，从而导致堆被耗尽。

但是，如果您将collect()替换为reduce()--就不再是问题了，因为后者将减少并丢弃通过的所有数据。

在所呈现的示例中，只需用reduce替换collect():

.reduce(0L, (aLong, result) -> aLong, (aLong1, aLong2) -> aLong1);

您甚至不需要考虑让计算依赖于 result ，因为Java不是一种纯FP（函数式编程）语言，无法优化流底部未使用的数据，因为可能会产生副作用。

- averasko

3

如果您不关心数据库保存的结果，您应该使用forEach...而不需要使用reduce。除非这是为了说明目的。 - DaveEdelstein

3

以下是代码示例

List<Integer> list = Arrays.asList(1,2,3,4,5,6,7);
int sum = list.stream().reduce((x,y) -> {
        System.out.println(String.format("x=%d,y=%d",x,y));
        return (x + y);
    }).get();

这里是执行结果：

System.out.println(sum);

x=1,y=2
x=3,y=3
x=6,y=4
x=10,y=5
x=15,y=6
x=21,y=7
28

Reduce函数接受两个参数，第一个参数是流中先前计算的返回值（整数），第二个参数是流中当前要计算的值。它将这两个值相加，并将结果作为下一次计算的第一个值。

- JetQin

1

始终优先使用collect()而不是reduce()方法，有非常充分的理由。在这里解释了，使用collect()可以提供更高效的性能。

Java 8 tutorial

*可变归约操作（例如Stream.collect()）在处理流元素时将其收集到可变结果容器（集合）中。与不可变归约操作（例如Stream.reduce()）相比，可变归约操作提供了更好的性能。

这是因为对于Collector，每次缩减步骤的结果都在可变的集合中，可以在下一步中再次使用该集合。

另一方面，Stream.reduce()操作使用不可变的结果容器，在每个中间步骤都需要实例化容器的新实例，从而降低性能。*

- Alex Mi

1

根据文档，

当在groupingBy或partitioningBy之后进行多级规约时，使用reducing()收集器最为有用。要在流上执行简单的规约，请改用Stream.reduce(BinaryOperator)。

因此，基本上只有在强制使用collect时才会使用reducing()。以下是另一个示例：

 For example, given a stream of Person, to calculate the longest last name 
 of residents in each city:

    Comparator<String> byLength = Comparator.comparing(String::length);
    Map<String, String> longestLastNameByCity
        = personList.stream().collect(groupingBy(Person::getCity,
            reducing("", Person::getLastName, BinaryOperator.maxBy(byLength))));

根据本教程，reduce有时效率较低。

reduce操作总是返回一个新值。然而，累加器函数每次处理流的元素时也返回一个新值。假设您想将流的元素缩减为更复杂的对象（例如集合），这可能会影响应用程序的性能。如果您的reduce操作涉及将元素添加到集合中，则每次累加器函数处理一个元素时，它都会创建一个包含该元素的新集合，这是低效的。最好更新现有的集合。您可以使用Stream.collect方法来实现这一点，下一节将介绍...

因此，在reduce场景中，“identity”是“可重复使用的”，因此如果可能，使用.reduce会稍微更有效率。

- rogerdpack

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Boris the Spider · Accepted Answer

reduce 是一种 "fold" 操作，它将二元运算符应用于流中的每个元素，其中运算符的第一个参数是上一次应用的返回值，第二个参数是当前流元素。

collect 是一种聚合操作，其中创建一个 "collection" 并将每个元素 "添加" 到该 collection 中。然后将来自流不同部分的 collections 加在一起。

document you linked 给出了使用两种不同方法的原因：

If we wanted to take a stream of strings and concatenate them into a single long string, we could achieve this with ordinary reduction:
 String concatenated = strings.reduce("", String::concat)  
We would get the desired result, and it would even work in parallel. However, we might not be happy about the performance! Such an implementation would do a great deal of string copying, and the run time would be O(n^2) in the number of characters. A more performant approach would be to accumulate the results into a StringBuilder, which is a mutable container for accumulating strings. We can use the same technique to parallelize mutable reduction as we do with ordinary reduction.

因此，要点在于并行化在两种情况下都是相同的，但在reduce情况下，我们将函数应用于流元素本身。在collect情况下，我们将函数应用于可变容器。

Java 8流 - collect与reduce

使用不可变结果对象的 "reduce()" 示例

"collect() with mutable" 示例

使用不可变结果对象的 "`reduce()`" 示例

"`collect()` with mutable" 示例