Java 8 Stream API：声明式数据处理与并行计算完全指南

Stream 核心概念

Stream 代表对象序列，支持声明式聚合操作（类似 SQL），特性：

• 元素序列：按需计算/获取，不存储数据。
• 来源：Collection、数组、I/O 通道、生成器。
• 聚合操作：filter/map/limit/reduce/find/match 等。
• 流水线：中间操作返回流，终端操作（如 collect）触发计算。
• 内部迭代：无需显式循环，自动利用多核（并行流）。

流创建

// Collection
List list = Arrays.asList("a", "b", "c");
Stream stream = list.stream();           // 顺序流
Stream pStream = list.parallelStream();  // 并行流

// 数组
String[] arr = {"a", "b"};
Stream aStream = Arrays.stream(arr);

// 静态工厂
Stream.of("a", "b", "c");
Stream.iterate(0, n -> n + 2).limit(5);  // 0,2,4,6,8
Stream.generate(Math::random).limit(3);  // 随机数

核心操作详解

forEach - 终端遍历

Random random = new Random();
random.ints().limit(10).forEach(System.out::println);

map - 映射变换

List numbers = Arrays.asList(3, 2, 2, 3, 7, 3, 5);
List squares = numbers.stream()
    .map(i -> i * i)
    .distinct()
    .collect(Collectors.toList());  // [9, 4, 49, 25]

filter - 过滤

List strings = Arrays.asList("abc", "", "bc", "efg", "abcd", "", "jkl");
long emptyCount = strings.stream().filter(String::isEmpty).count();  // 2

limit - 截取

random.ints().limit(10).forEach(System.out::println);

sorted - 排序

random.ints().limit(10).sorted().forEach(System.out::println);

collect - 收集

List filtered = strings.stream()
    .filter(s -> !s.isEmpty())
    .collect(Collectors.toList());

String merged = strings.stream()
    .filter(s -> !s.isEmpty())
    .collect(Collectors.joining(", "));  // "abc, bc, efg, abcd, jkl"

统计收集器

List numbers = Arrays.asList(3, 2, 2, 3, 7, 3, 5);
IntSummaryStatistics stats = numbers.stream()
    .mapToInt(x -> x)
    .summaryStatistics();

System.out.println("Max: " + stats.getMax());      // 7
System.out.println("Min: " + stats.getMin());      // 2
System.out.println("Sum: " + stats.getSum());      // 25
System.out.println("Avg: " + stats.getAverage());  // 3.57...

并行流

long count = strings.parallelStream()
    .filter(String::isEmpty)
    .count();

仅将 stream() 改为 parallelStream() 即可并行化，ForkJoinPool 管理线程。注意：有状态操作（如 distinct、sorted）并行收益有限；小数据量并行开销可能大于收益。

完整对比示例：Java 7 vs Java 8

public class Java8Tester {
    public static void main(String[] args) {
        List strings = Arrays.asList("abc", "", "bc", "efg", "abcd", "", "jkl");
        List numbers = Arrays.asList(3, 2, 2, 3, 7, 3, 5);
        List integers = Arrays.asList(1,2,13,4,15,6,17,8,19);

        System.out.println("=== Java 7 方式 ===");
        System.out.println("Empty: " + countEmpty7(strings));
        System.out.println("Len 3: " + countLen3_7(strings));
        System.out.println("Filtered: " + filterEmpty7(strings));
        System.out.println("Merged: " + merge7(strings));
        System.out.println("Squares: " + squares7(numbers));
        printStats7(integers);
        printRandom7();

        System.out.println("\n=== Java 8 Stream 方式 ===");
        System.out.println("Empty: " + strings.stream().filter(String::isEmpty).count());
        System.out.println("Len 3: " + strings.stream().filter(s -> s.length()==3).count());
        System.out.println("Filtered: " + strings.stream().filter(s -> !s.isEmpty()).collect(Collectors.toList()));
        System.out.println("Merged: " + strings.stream().filter(s -> !s.isEmpty()).collect(Collectors.joining(", ")));
        System.out.println("Squares: " + numbers.stream().map(i -> i*i).distinct().collect(Collectors.toList()));
        IntSummaryStatistics stats = integers.stream().mapToInt(x -> x).summaryStatistics();
        System.out.println("Max: " + stats.getMax() + " Min: " + stats.getMin() + " Sum: " + stats.getSum() + " Avg: " + stats.getAverage());
        new Random().ints().limit(10).sorted().forEach(System.out::println);
        System.out.println("Parallel empty: " + strings.parallelStream().filter(String::isEmpty).count());
    }

    // Java 7 实现略（含循环、判断、StringBuilder 等样板代码）
}

最佳实践

• 优先使用方法引用（String::isEmpty）替代简单 Lambda。
• 避免在流管道中产生副作用（修改外部状态），保持纯函数风格。
• 大数据量、CPU 密集型操作考虑并行流；IO 密集或小数据用顺序流。
• 使用 mapToInt/Long/Double 避免装箱开销，配合原语流专用收集器。
• 复杂逻辑拆分为多个流管道或使用 Collector 自定义收集器。