Java Stream 流终止后复用错误的原因
2022-05-238.0k 阅读
Java Stream 流的基本概念
在深入探讨 Java Stream 流终止后复用错误的原因之前，我们先来回顾一下 Java Stream 的基本概念。Java 8 引入的 Stream API 为处理集合数据提供了一种更为简洁和高效的方式。Stream 代表一系列元素，这些元素来自数据源（如集合、数组等），并且支持各种操作，如过滤、映射、排序和聚合等。
Stream 操作主要分为两类：中间操作和终止操作。中间操作返回一个新的 Stream，允许链式调用多个操作，例如 filter、map 等方法，它们是惰性求值的，即只有在终止操作被调用时才会真正执行。而终止操作则会触发流的处理，并返回一个非流的结果，比如 forEach、collect、count 等方法。
代码示例理解基本操作
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
public class StreamBasicExample {
public static void main(String[] args) {
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
// 中间操作：过滤出偶数，映射为平方
// 这一步并不会立即执行过滤和映射操作，只是构建操作链
var stream = numbers.stream()
.filter(n -> n % 2 == 0)
.map(n -> n * n);
// 终止操作：收集结果为列表
// 此时才会真正执行前面构建的操作链
List<Integer> result = stream.collect(Collectors.toList());
System.out.println(result);
}
}
在上述代码中，stream 首先通过 filter 方法过滤出偶数，再通过 map 方法将每个偶数平方。但这些操作并没有立即执行，直到 collect 终止操作被调用，才会触发整个操作链的执行，最终得到结果列表 [4, 16]。
Stream 流终止后复用问题的引出
当我们尝试在一个 Stream 执行终止操作后再次复用它时，就会遇到问题。例如，以下代码：
import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;
public class StreamReuseExample {
public static void main(String[] args) {
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
Stream<Integer> stream = numbers.stream();
// 第一次终止操作：打印流中的元素
stream.forEach(System.out::println);
// 尝试再次复用流
stream.forEach(System.out::println);
}
}
运行这段代码会抛出 IllegalStateException 异常，提示流已经被操作或关闭。这表明 Stream 在执行一次终止操作后，就不能再被复用了。接下来我们深入分析其背后的原因。
从 Stream 的实现原理分析复用错误原因
- Stream 的状态和生命周期 Stream 并不是一个数据结构，它不存储元素，而是在操作过程中按需从数据源获取元素。每个 Stream 实例都有自己的状态，包括数据源、中间操作链以及是否已经被操作或关闭的标志。当一个 Stream 执行终止操作时，它的状态会发生变化，其中一个关键变化就是将“已操作”标志设为 true。
// 简化的 Stream 实现类结构示意
abstract class AbstractStream<T, S extends Stream<T>> {
private boolean isOperated;
// 中间操作方法示意
S intermediateOperation() {
// 构建操作链等逻辑
return (S) this;
}
// 终止操作方法示意
Object terminalOperation() {
if (isOperated) {
throw new IllegalStateException("Stream has already been operated upon or closed");
}
isOperated = true;
// 执行终止操作逻辑
return result;
}
}
- 数据源的消耗 Stream 依赖数据源来提供元素。在执行终止操作过程中，数据源的元素会被“消耗”。例如，对于基于集合的 Stream，其内部实现可能是通过迭代器来遍历集合元素。当执行终止操作时，迭代器会逐步遍历集合，直到操作完成。如果再次尝试复用该 Stream，数据源已经处于遍历完成的状态，无法再次提供元素。
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
public class StreamDataSourceConsumption {
public static void main(String[] args) {
List<Integer> list = new ArrayList<>();
list.add(1);
list.add(2);
list.add(3);
Iterator<Integer> iterator = list.iterator();
while (iterator.hasNext()) {
System.out.println(iterator.next());
}
// 此时 iterator 已经遍历完列表，再次调用 hasNext() 将返回 false
System.out.println("再次尝试遍历：" + (iterator.hasNext()? iterator.next() : "无元素"));
}
}
在上述代码中，通过迭代器遍历列表后，迭代器已经处于遍历完成状态，再次尝试获取元素就无法得到新的结果。Stream 在执行终止操作时，类似地消耗了数据源，导致复用流时无数据可用。
- 操作链的重置问题 Stream 的中间操作会构建一个操作链，在终止操作执行时，这个操作链会被一次性执行。一旦终止操作完成，操作链就已经执行完毕，无法自动重置。如果再次复用流，就相当于要重新执行已经执行过的操作链，但 Stream 并没有提供这样的机制来重置操作链。
import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;
public class StreamOperationChainReset {
public static void main(String[] args) {
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
Stream<Integer> stream = numbers.stream()
.filter(n -> n % 2 == 0)
.map(n -> n * n);
// 第一次终止操作
stream.count();
// 尝试再次复用流，操作链已执行，无法重置
stream.forEach(System.out::println);
}
}
在这个例子中，filter 和 map 构建了操作链，count 终止操作执行后，操作链已执行完毕，再次尝试复用流进行 forEach 操作就会失败，因为操作链不能自动重置。
深入探究 Stream 的内部状态管理
- 内部迭代与外部迭代
传统的集合遍历是外部迭代，例如使用
for循环或者迭代器，开发者需要显式地控制迭代过程。而 Stream 采用的是内部迭代，开发者只需要声明要执行的操作，Stream 内部负责迭代数据源并执行操作链。在内部迭代过程中，Stream 会维护自身的状态，包括当前处理到数据源的哪个位置等。当终止操作完成后，这些状态已经处于完成状态，无法恢复到初始状态以便再次复用。
import java.util.ArrayList;
import java.util.List;
// 外部迭代示例
public class ExternalIteration {
public static void main(String[] args) {
List<Integer> numbers = new ArrayList<>();
numbers.add(1);
numbers.add(2);
numbers.add(3);
for (int number : numbers) {
System.out.println(number);
}
}
}
// 内部迭代示例
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Stream;
public class InternalIteration {
public static void main(String[] args) {
List<Integer> numbers = new ArrayList<>();
numbers.add(1);
numbers.add(2);
numbers.add(3);
Stream<Integer> stream = numbers.stream();
stream.forEach(System.out::println);
}
}
在外部迭代中，每次使用 for 循环都可以重新开始遍历集合。而内部迭代的 Stream 在执行终止操作后，其内部状态已改变，无法重新开始类似的遍历操作。
- 状态标识位的作用 Stream 内部通过一些标识位来管理自身状态，除了前面提到的“已操作”标识位，还有可能涉及到“是否并行”等标识位。这些标识位在 Stream 的生命周期中起着重要作用，它们决定了 Stream 在不同阶段的行为。当终止操作执行后，这些标识位会被设置成特定状态，使得 Stream 处于不可复用状态。
// 简化的 Stream 状态标识示意
abstract class AbstractStream<T, S extends Stream<T>> {
private boolean isOperated;
private boolean isParallel;
// 设置并行状态的方法
S parallel() {
isParallel = true;
return (S) this;
}
// 终止操作方法示意
Object terminalOperation() {
if (isOperated) {
throw new IllegalStateException("Stream has already been operated upon or closed");
}
isOperated = true;
// 根据 isParallel 等状态执行不同的终止操作逻辑
return result;
}
}
例如，parallel 方法设置了并行标识位，在终止操作时，Stream 会根据这个标识位决定是否采用并行计算策略。一旦终止操作完成，标识位状态固定，Stream 也就不能像初始状态那样被复用。
从 JVM 角度分析 Stream 复用错误
- 内存管理与 Stream 状态 从 JVM 的内存管理角度来看，Stream 实例在内存中占用一定的空间，并且其状态信息也存储在内存中。当执行终止操作后，Stream 实例的状态信息发生改变，这些改变后的状态会一直保留在内存中。如果尝试复用 Stream，JVM 并不会自动将这些状态恢复到初始状态，因为这不符合 Stream 的设计初衷。Stream 的设计是为了一次操作完成特定任务，而不是重复使用。
- 对象生命周期与 Stream 复用 在 JVM 中，对象有其自身的生命周期。Stream 实例在创建后，随着操作的进行，其生命周期会发生变化。执行终止操作可以看作是 Stream 生命周期的一个重要阶段，在这个阶段之后，Stream 已经完成了其设计的任务，从某种意义上说，它已经“用过”了。再次复用它就类似于试图延长一个已经完成任务的对象的生命周期，这与 JVM 对对象生命周期的管理机制不相符。
解决 Stream 复用问题的方法
- 重新创建 Stream 最直接的解决方法就是在需要再次使用流操作时，重新创建 Stream。例如：
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
public class StreamRecreationExample {
public static void main(String[] args) {
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
// 第一次操作
List<Integer> result1 = numbers.stream()
.filter(n -> n % 2 == 0)
.collect(Collectors.toList());
System.out.println(result1);
// 第二次操作，重新创建 Stream
List<Integer> result2 = numbers.stream()
.map(n -> n * 10)
.collect(Collectors.toList());
System.out.println(result2);
}
}
在上述代码中，每次需要进行流操作时，都通过 numbers.stream() 重新创建 Stream，这样就可以避免复用已终止流带来的问题。
- 使用 Supplier 来延迟创建 Stream
可以使用
Supplier来延迟 Stream 的创建，从而在需要时能够方便地获取新的 Stream。
import java.util.Arrays;
import java.util.List;
import java.util.function.Supplier;
import java.util.stream.Collectors;
public class StreamSupplierExample {
public static void main(String[] args) {
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
Supplier<java.util.stream.Stream<Integer>> streamSupplier = () -> numbers.stream();
// 第一次操作
List<Integer> result1 = streamSupplier.get()
.filter(n -> n % 2 == 0)
.collect(Collectors.toList());
System.out.println(result1);
// 第二次操作，通过 Supplier 获取新的 Stream
List<Integer> result2 = streamSupplier.get()
.map(n -> n * 10)
.collect(Collectors.toList());
System.out.println(result2);
}
}
通过 Supplier，我们将 Stream 的创建逻辑封装起来，每次调用 get 方法都能获取一个新的 Stream，避免了流复用错误。
总结 Stream 复用错误的本质
Java Stream 流终止后复用错误的本质在于 Stream 的设计理念和实现机制。Stream 采用内部迭代，在执行终止操作时会改变自身状态，消耗数据源，并执行操作链，这些改变使得 Stream 不能像初始状态那样再次被使用。从 JVM 角度看，其内存管理和对象生命周期机制也不支持 Stream 的复用。理解这些本质原因，有助于开发者在使用 Stream 时避免因复用已终止流而导致的错误，正确地利用 Stream API 高效处理集合数据。同时，掌握重新创建 Stream 或使用 Supplier 等解决方法，可以让开发者在需要多次进行流操作时，写出更加健壮和正确的代码。通过深入理解和实践，开发者能够更好地驾驭 Java Stream 这一强大的工具，提升代码的质量和效率。