## 概述 Java8 中引入的 Stream API 是一个强大的函数式编程工具，它允许您以声明式的方式处理数据集合。Stream 不是数据结构，而是对数据源的一种抽象，提供了高效、简洁的数据处理方式。 ### Stream 的核心特性 - **声明式编程**: 描述做什么，而不是怎么做 - **函数式编程**: 支持链式调用和无副作用的操作 - **惰性求值**: 中间操作只有在终端操作时才会执行 - **内部迭代**: 由 Stream API 控制迭代过程 ## 创建 Stream ### 1. 从集合创建 ```java // 从 List 创建 List list = Arrays.asList("Java", "Python", "JavaScript"); Stream stream = list.stream(); // 从数组创建 String[] array = {"Java", "Python", "JavaScript"}; Stream stream = Arrays.stream(array); // 从 Set 创建 Set set = new HashSet<>(Arrays.asList(1, 2, 3, 4, 5)); Stream stream = set.stream(); ``` ### 2. 使用 Stream.of() ```java // 创建指定元素的流 Stream stream = Stream.of("Java", "Python", "JavaScript"); // 创建空流 Stream emptyStream = Stream.empty(); ``` ### 3. 无限流 ```java // 生成无限流 - iterate Stream evenNumbers = Stream.iterate(0, n -> n + 2); // 生成无限流 - generate Stream randomNumbers = Stream.generate(Math::random); // 指定范围的数字流 IntStream range = IntStream.range(1, 11); // 1 到 10 IntStream rangeClosed = IntStream.rangeClosed(1, 10); // 1 到 10（包含10） ``` ### 4. 从文件创建 ```java try { // 读取文件行 Stream lines = Files.lines(Paths.get("file.txt")); // 读取目录中的文件 Stream files = Files.list(Paths.get("directory")); } catch (IOException e) { e.printStackTrace(); } ``` ## 中间操作 (Intermediate Operations) 中间操作返回一个新的 Stream，可以进行链式调用。这些操作是惰性的，只有在终端操作调用时才会执行。 ### 1. filter() - 过滤 ```java List numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10); // 过滤偶数 List evenNumbers = numbers.stream() .filter(n -> n % 2 == 0) .collect(Collectors.toList()); // 结果: [2, 4, 6, 8, 10] // 过滤字符串长度大于5的元素 List languages = Arrays.asList("Java", "Python", "JavaScript", "Go", "Rust"); List longNames = languages.stream() .filter(name -> name.length() > 5) .collect(Collectors.toList()); // 结果: ["Python", "JavaScript"] ``` ### 2. map() - 映射转换 ```java List words = Arrays.asList("hello", "world", "java", "stream"); // 转换为大写 List upperCase = words.stream() .map(String::toUpperCase) .collect(Collectors.toList()); // 结果: ["HELLO", "WORLD", "JAVA", "STREAM"] // 获取字符串长度 List lengths = words.stream() .map(String::length) .collect(Collectors.toList()); // 结果: [5, 5, 4, 6] // 对象映射 List persons = Arrays.asList( new Person("Alice", 25), new Person("Bob", 30) ); List names = persons.stream() .map(Person::getName) .collect(Collectors.toList()); ``` ### 3. flatMap() - 扁平化映射 ```java List> nestedList = Arrays.asList( Arrays.asList("Java", "Python"), Arrays.asList("JavaScript", "TypeScript"), Arrays.asList("Go", "Rust") ); // 扁平化为单一列表 List flatList = nestedList.stream() .flatMap(Collection::stream) .collect(Collectors.toList()); // 结果: ["Java", "Python", "JavaScript", "TypeScript", "Go", "Rust"] // 字符串拆分并扁平化 List sentences = Arrays.asList("Hello World", "Java Stream API"); List words = sentences.stream() .flatMap(sentence -> Arrays.stream(sentence.split(" "))) .collect(Collectors.toList()); // 结果: ["Hello", "World", "Java", "Stream", "API"] ``` ### 4. distinct() - 去重 ```java List numbers = Arrays.asList(1, 2, 2, 3, 3, 4, 5, 5); List distinctNumbers = numbers.stream() .distinct() .collect(Collectors.toList()); // 结果: [1, 2, 3, 4, 5] // 对象去重（需要重写 equals 和 hashCode） List uniquePersons = persons.stream() .distinct() .collect(Collectors.toList()); ``` ### 5. sorted() - 排序 ```java List numbers = Arrays.asList(3, 1, 4, 1, 5, 9, 2, 6); // 自然排序 List sorted = numbers.stream() .sorted() .collect(Collectors.toList()); // 结果: [1, 1, 2, 3, 4, 5, 6, 9] // 自定义排序 List words = Arrays.asList("apple", "banana", "cherry", "date"); List sortedByLength = words.stream() .sorted(Comparator.comparing(String::length)) .collect(Collectors.toList()); // 结果: ["date", "apple", "banana", "cherry"] // 倒序排序 List descending = numbers.stream() .sorted(Comparator.reverseOrder()) .collect(Collectors.toList()); // 多字段排序 List sortedPersons = persons.stream() .sorted(Comparator.comparing(Person::getAge) .thenComparing(Person::getName)) .collect(Collectors.toList()); ``` ### 6. limit() 和 skip() - 限制和跳过 ```java List numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10); // 取前5个元素 List first5 = numbers.stream() .limit(5) .collect(Collectors.toList()); // 结果: [1, 2, 3, 4, 5] // 跳过前3个元素 List skipFirst3 = numbers.stream() .skip(3) .collect(Collectors.toList()); // 结果: [4, 5, 6, 7, 8, 9, 10] // 分页：跳过前5个，取接下来的3个 List page = numbers.stream() .skip(5) .limit(3) .collect(Collectors.toList()); // 结果: [6, 7, 8] ``` ### 7. peek() - 调试和副作用 ```java List words = Arrays.asList("hello", "world", "java"); List result = words.stream() .peek(word -> System.out.println("Processing: " + word)) .map(String::toUpperCase) .peek(word -> System.out.println("Transformed: " + word)) .collect(Collectors.toList()); ``` ## 终端操作 (Terminal Operations) 终端操作会触发流的执行，并产生最终结果。每个流只能有一个终端操作。 ### 1. collect() - 收集器 ```java List words = Arrays.asList("java", "python", "javascript", "go"); // 收集到 List List list = words.stream() .filter(w -> w.length() > 4) .collect(Collectors.toList()); // 收集到 Set Set set = words.stream() .collect(Collectors.toSet()); // 收集到 Map Map wordLengths = words.stream() .collect(Collectors.toMap( word -> word, // key String::length // value )); // 分组 Map> groupedByLength = words.stream() .collect(Collectors.groupingBy(String::length)); // 分区 Map> partitioned = words.stream() .collect(Collectors.partitioningBy(w -> w.length() > 4)); // 连接字符串 String joined = words.stream() .collect(Collectors.joining(", ")); // 结果: "java, python, javascript, go" String joinedWithPrefixSuffix = words.stream() .collect(Collectors.joining(", ", "[", "]")); // 结果: "[java, python, javascript, go]" ``` ### 2. reduce() - 规约操作 ```java List numbers = Arrays.asList(1, 2, 3, 4, 5); // 求和 Optional sum = numbers.stream() .reduce(Integer::sum); // 或者提供初始值 Integer sum2 = numbers.stream() .reduce(0, Integer::sum); // 求最大值 Optional max = numbers.stream() .reduce(Integer::max); // 字符串连接 List words = Arrays.asList("Hello", " ", "World"); Optional concatenated = words.stream() .reduce(String::concat); // 复杂规约：计算平均值 OptionalDouble average = numbers.stream() .mapToInt(Integer::intValue) .average(); ``` ### 3. forEach() - 遍历 ```java List words = Arrays.asList("java", "python", "javascript"); // 简单遍历 words.stream() .forEach(System.out::println); // 带索引的遍历 IntStream.range(0, words.size()) .forEach(i -> System.out.println(i + ": " + words.get(i))); ``` ### 4. 查找操作 ```java List numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10); // 查找第一个偶数 Optional firstEven = numbers.stream() .filter(n -> n % 2 == 0) .findFirst(); // 查找任意一个偶数（并行流中有用） Optional anyEven = numbers.stream() .filter(n -> n % 2 == 0) .findAny(); // 检查是否所有元素都满足条件 boolean allPositive = numbers.stream() .allMatch(n -> n > 0); // 检查是否任意元素满足条件 boolean hasEven = numbers.stream() .anyMatch(n -> n % 2 == 0); // 检查是否所有元素都不满足条件 boolean noneNegative = numbers.stream() .noneMatch(n -> n < 0); ``` ### 5. 统计操作 ```java List numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10); // 计数 long count = numbers.stream() .filter(n -> n % 2 == 0) .count(); // 使用数值流进行统计 IntSummaryStatistics stats = numbers.stream() .mapToInt(Integer::intValue) .summaryStatistics(); System.out.println("Count: " + stats.getCount()); System.out.println("Sum: " + stats.getSum()); System.out.println("Min: " + stats.getMin()); System.out.println("Max: " + stats.getMax()); System.out.println("Average: " + stats.getAverage()); ``` ## 数值流 (Primitive Streams) Java8 提供了专门的数值流来避免装箱/拆箱的性能损耗。 ### IntStream、LongStream、DoubleStream ```java // 创建数值流 IntStream intStream = IntStream.range(1, 11); LongStream longStream = LongStream.rangeClosed(1, 100); DoubleStream doubleStream = DoubleStream.of(1.1, 2.2, 3.3); // 统计操作 int sum = IntStream.range(1, 11).sum(); OptionalDouble average = IntStream.range(1, 11).average(); OptionalInt max = IntStream.range(1, 11).max(); OptionalInt min = IntStream.range(1, 11).min(); // 转换为流 Stream boxed = IntStream.range(1, 11).boxed(); // 从普通流转换为数值流 List numbers = Arrays.asList("1", "2", "3", "4", "5"); IntStream intStream2 = numbers.stream() .mapToInt(Integer::parseInt); ``` ## 并行流 (Parallel Streams) 并行流可以利用多核处理器的优势，但需要谨慎使用。 ### 创建并行流 ```java List numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10); // 从集合创建并行流 Stream parallelStream = numbers.parallelStream(); // 将顺序流转换为并行流 Stream parallel = numbers.stream().parallel(); // 检查是否为并行流 boolean isParallel = numbers.parallelStream().isParallel(); // 转换回顺序流 Stream sequential = numbers.parallelStream().sequential(); ``` ### 并行流示例 ```java // 并行计算大数据集的和 List largeList = IntStream.range(1, 1000000) .boxed() .collect(Collectors.toList()); // 顺序流 long startTime = System.currentTimeMillis(); long sum1 = largeList.stream() .mapToLong(Integer::longValue) .sum(); long sequentialTime = System.currentTimeMillis() - startTime; // 并行流 startTime = System.currentTimeMillis(); long sum2 = largeList.parallelStream() .mapToLong(Integer::longValue) .sum(); long parallelTime = System.currentTimeMillis() - startTime; System.out.println("顺序流耗时: " + sequentialTime + "ms"); System.out.println("并行流耗时: " + parallelTime + "ms"); ``` ### 并行流注意事项 ```java // ❌ 错误：在并行流中使用非线程安全的操作 List list = new ArrayList<>(); // ArrayList 不是线程安全的 IntStream.range(1, 1000) .parallel() .forEach(list::add); // 可能导致数据丢失或异常 // ✅ 正确：使用线程安全的收集器 List safeList = IntStream.range(1, 1000) .parallel() .boxed() .collect(Collectors.toList()); // ✅ 正确：使用线程安全的集合 List synchronizedList = Collections.synchronizedList(new ArrayList<>()); IntStream.range(1, 1000) .parallel() .forEach(synchronizedList::add); ``` ## 自定义收集器 ### 使用 Collector.of() 创建自定义收集器 ```java // 创建一个收集器，将字符串连接并添加括号 Collector parenthesesCollector = Collector.of( StringBuilder::new, // supplier: 创建容器 (sb, s) -> sb.append("(").append(s).append(")"), // accumulator: 累加元素 StringBuilder::append, // combiner: 合并容器（并行流用） StringBuilder::toString // finisher: 最终转换 ); List words = Arrays.asList("hello", "world", "java"); String result = words.stream() .collect(parenthesesCollector); // 结果: "(hello)(world)(java)" // 创建一个统计收集器 Collector averageCollector = Collector.of( () -> new int[2], // [sum, count] (acc, value) -> { acc[0] += value; acc[1]++; }, (acc1, acc2) -> { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; }, acc -> (double) acc[0] / acc[1] ); List numbers = Arrays.asList(1, 2, 3, 4, 5); Double average = numbers.stream() .collect(averageCollector); ``` ## 实际应用案例 ### 案例1：处理员工数据 ```java public class Employee { private String name; private String department; private double salary; private int age; // 构造器、getter、setter 省略 } List employees = Arrays.asList( new Employee("Alice", "IT", 70000, 25), new Employee("Bob", "Finance", 65000, 30), new Employee("Charlie", "IT", 80000, 28), new Employee("Diana", "HR", 60000, 35), new Employee("Eve", "IT", 75000, 27) ); // 1. 找出IT部门薪资最高的员工 Optional highestPaidIT = employees.stream() .filter(emp -> "IT".equals(emp.getDepartment())) .max(Comparator.comparing(Employee::getSalary)); // 2. 按部门分组并计算平均薪资 Map avgSalaryByDept = employees.stream() .collect(Collectors.groupingBy( Employee::getDepartment, Collectors.averagingDouble(Employee::getSalary) )); // 3. 找出每个部门年龄最小的员工 Map> youngestByDept = employees.stream() .collect(Collectors.groupingBy( Employee::getDepartment, Collectors.minBy(Comparator.comparing(Employee::getAge)) )); // 4. 薪资超过70000的员工姓名列表 List highEarners = employees.stream() .filter(emp -> emp.getSalary() > 70000) .map(Employee::getName) .sorted() .collect(Collectors.toList()); // 5. 计算总薪资支出 double totalSalary = employees.stream() .mapToDouble(Employee::getSalary) .sum(); ``` ### 案例2：文本处理 ```java // 处理文本文件，统计单词频率 public Map wordFrequency(String text) { return Arrays.stream(text.toLowerCase().split("\\W+")) .filter(word -> !word.isEmpty()) .collect(Collectors.groupingBy( Function.identity(), Collectors.counting() )); } // 查找最常用的前N个单词 public List> topWords(String text, int n) { return wordFrequency(text).entrySet().stream() .sorted(Map.Entry.comparingByValue().reversed()) .limit(n) .collect(Collectors.toList()); } // 使用示例 String text = "Java is great Java is powerful Java streams are amazing"; Map frequencies = wordFrequency(text); List> top3 = topWords(text, 3); ``` ### 案例3：数据转换和验证 ```java public class Order { private String id; private String customerId; private List items; private LocalDateTime orderDate; // 构造器、getter、setter 省略 } public class OrderItem { private String productId; private int quantity; private double price; // 构造器、getter、setter 省略 } List orders = getOrders(); // 获取订单列表 // 1. 计算每个客户的总消费 Map customerSpending = orders.stream() .collect(Collectors.groupingBy( Order::getCustomerId, Collectors.summingDouble(order -> order.getItems().stream() .mapToDouble(item -> item.getQuantity() * item.getPrice()) .sum() ) )); // 2. 找出本月订单数量超过10的产品 Map productOrderCount = orders.stream() .filter(order -> order.getOrderDate().getMonth() == LocalDateTime.now().getMonth()) .flatMap(order -> order.getItems().stream()) .collect(Collectors.groupingBy( OrderItem::getProductId, Collectors.summingLong(OrderItem::getQuantity) )) .entrySet().stream() .filter(entry -> entry.getValue() > 10) .collect(Collectors.toMap( Map.Entry::getKey, Map.Entry::getValue )); // 3. 验证订单数据 List invalidOrders = orders.stream() .filter(order -> order.getItems().isEmpty() || order.getCustomerId() == null || order.getItems().stream().anyMatch(item -> item.getPrice() < 0)) .map(Order::getId) .collect(Collectors.toList()); ``` ## 性能优化和最佳实践 ### 1. 选择合适的流类型 ```java // ✅ 对于数值计算，使用原始类型流 int sum = IntStream.range(1, 1000000) .sum(); // 比 Stream 更高效 // ❌ 避免不必要的装箱 Stream numbers = IntStream.range(1, 1000000) .boxed(); // 产生装箱开销 ``` ### 2. 合理使用并行流 ```java // ✅ 适合并行的场景：CPU密集型任务，大数据集 List results = largeDataSet.parallelStream() .map(this::complexCalculation) .collect(Collectors.toList()); // ❌ 不适合并行的场景：IO密集型任务，小数据集 List smallList = Arrays.asList("a", "b", "c"); smallList.parallelStream() // 并行开销可能超过收益 .map(String::toUpperCase) .collect(Collectors.toList()); ``` ### 3. 优化操作顺序 ```java List words = getHugeWordList(); // ✅ 优化：先过滤再排序 List result1 = words.stream() .filter(word -> word.length() > 10) // 先过滤，减少排序元素 .sorted() .collect(Collectors.toList()); // ❌ 低效：先排序再过滤 List result2 = words.stream() .sorted() // 对所有元素排序 .filter(word -> word.length() > 10) // 然后过滤 .collect(Collectors.toList()); ``` ### 4. 避免重复计算 ```java // ❌ 重复计算 List expensiveResults = words.stream() .filter(word -> expensiveOperation(word) > 100) .map(word -> expensiveOperation(word)) // 重复调用 .collect(Collectors.toList()); // ✅ 缓存计算结果 List efficientResults = words.stream() .map(word -> new AbstractMap.SimpleEntry<>(word, expensiveOperation(word))) .filter(entry -> entry.getValue() > 100) .map(Map.Entry::getKey) .collect(Collectors.toList()); ``` ### 5. 合理使用 Short-circuit 操作 ```java // ✅ 使用 findFirst() 而不是完整遍历 Optional found = hugeList.stream() .filter(this::complexPredicate) .findFirst(); // 找到第一个就停止 // ✅ 使用 anyMatch() 检查条件 boolean hasMatch = hugeList.stream() .anyMatch(this::complexPredicate); // 找到就返回 true ``` ## 常见陷阱和注意事项 ### 1. 流只能使用一次 ```java Stream stream = Arrays.asList("a", "b", "c").stream(); stream.forEach(System.out::println); // 正常 // stream.count(); // ❌ IllegalStateException: 流已经被使用 ``` ### 2. 副作用操作 ```java List results = new ArrayList<>(); // ❌ 在流操作中产生副作用 words.stream() .filter(word -> word.length() > 3) .forEach(results::add); // 副作用操作 // ✅ 使用收集器 List results2 = words.stream() .filter(word -> word.length() > 3) .collect(Collectors.toList()); ``` ### 3. 并行流的状态共享 ```java int[] counter = {0}; // ❌ 并行流中的竞态条件 IntStream.range(1, 1000) .parallel() .forEach(i -> counter[0]++); // 不安全 // ✅ 使用原子操作或收集器 AtomicInteger atomicCounter = new AtomicInteger(0); IntStream.range(1, 1000) .parallel() .forEach(i -> atomicCounter.incrementAndGet()); ``` ### 4. 正确处理 Optional ```java // ❌ 不当使用 Optional String result = optionalString.isPresent() ? optionalString.get() : "default"; // ✅ 使用 Optional 的方法 String result2 = optionalString.orElse("default"); String result3 = optionalString.orElseGet(() -> getDefaultValue()); ``` ## 总结 Java8 Stream API 提供了强大而简洁的数据处理能力，主要优势包括： 1. **提高代码可读性**: 声明式编程风格使代码更容易理解 2. **减少样板代码**: 避免显式的循环和条件判断 3. **支持并行处理**: 轻松利用多核处理器的优势 4. **函数式编程**: 支持链式调用和组合操作 5. **惰性求值**: 提高性能，避免不必要的计算 ### 使用建议 - 对于简单的集合操作，优先考虑使用 Stream API - 注意性能影响，特别是在处理大数据集时 - 合理使用并行流，避免在不合适的场景下使用 - 保持操作的无副作用性，使用纯函数 - 利用 IDE 的重构功能，将传统循环转换为 Stream 操作掌握 Stream API 将极大提升您的 Java 开发效率和代码质量。建议在实际项目中多加练习，逐步熟悉各种操作的使用场景和性能特征。

Java8 流式开发指南