Java真的不难（五十五）Stream流的进阶用法
Stream的高级特性
Java 8 引入的 Stream API 提供了强大的数据处理能力，包括中间操作、终端操作、收集器和并行流等高级特性，让我们一起来感受它的魅力~
1、流的中间操作与终端操作
Java Stream的中间操作（如filter、map）是惰性的，它们返回一个新流并构建处理流水线，但不会立即执行。只有调用终端操作（如collect、forEach）时，整个流才会被处理并触发实际计算。中间操作可链式调用，而终端操作会消耗流，使其不可复用。例如，stream.map(…).filter(…).collect(…)中，map和filter是中间操作，collect是终端操作。
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;public class StreamOperations {public static void main(String[] args) {List<String> names = Arrays.asList("Alice", "Bob", "Charlie", "David", "Eve", "Frank");// 链式流操作：过滤 -> 映射 -> 排序 -> 收集List<String> processedNames = names.stream().filter(name -> name.length() > 3) // 中间操作：过滤长度>3的名字.map(String::toUpperCase) // 中间操作：转为大写.sorted((a, b) -> b.compareTo(a)) // 中间操作：逆序排序.collect(Collectors.toList()); // 终端操作：收集为ListSystem.out.println("Processed names: " + processedNames);// 终端操作：forEachSystem.out.print("Printing names: ");names.stream().limit(4) // 中间操作：限制前4个元素.forEach(name -> System.out.print(name + " ")); // 终端操作：遍历打印// 终端操作：reduceString concatenated = names.stream().reduce("", (a, b) -> a + "|" + b); // 将字符串用|连接System.out.println("\nConcatenated: " + concatenated);}
}
2、高级收集器
Collectors类提供了高级归约操作，如分组（groupingBy）、分区（partitioningBy）、拼接字符串（joining）等。这些收集器可将流元素转换为集合、Map或统计结果。例如，groupingBy按属性分组，toMap将流转为键值对。自定义组合收集器还能实现复杂聚合，如多级分组或下游聚合（如counting()）。
import java.util.*;
import java.util.stream.Collectors;public class AdvancedCollectors {public static void main(String[] args) {List<Person> people = Arrays.asList(new Person("Alice", 25, "New York"),new Person("Bob", 30, "London"),new Person("Charlie", 20, "New York"),new Person("David", 35, "London"),new Person("Eve", 28, "Paris"));// 按城市分组Map<String, List<Person>> peopleByCity = people.stream().collect(Collectors.groupingBy(Person::getCity));System.out.println("People by city: " + peopleByCity);// 按城市分组并计算每组的平均年龄Map<String, Double> avgAgeByCity = people.stream().collect(Collectors.groupingBy(Person::getCity,Collectors.averagingInt(Person::getAge)));System.out.println("Average age by city: " + avgAgeByCity);// 分区：按年龄是否>=30分为两组Map<Boolean, List<Person>> partitioned = people.stream().collect(Collectors.partitioningBy(p -> p.getAge() >= 30));System.out.println("Partitioned by age >=30: " + partitioned);// 连接字符串String allNames = people.stream().map(Person::getName).collect(Collectors.joining(", ", "[", "]"));System.out.println("All names: " + allNames);}
}class Person {private String name;private int age;private String city;// 构造方法、getter和setter省略...
}
3、 并行流与性能考虑
并行流（ParallelStream()）利用多核处理器拆分任务并行处理，但需注意线程安全、任务开销和数据特性。适合大规模数据且无状态的操作，但可能因线程竞争或频繁拆箱/装箱降低性能。避免共享可变状态，并考虑Spliterator实现的分割效率。基准测试是评估并行效果的关键。
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ForkJoinPool;public class ParallelStreams {public static void main(String[] args) {List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);// 顺序流long start = System.currentTimeMillis();int sumSequential = numbers.stream().mapToInt(ParallelStreams::slowSquare) // 模拟耗时操作.sum();long duration = System.currentTimeMillis() - start;System.out.println("Sequential sum: " + sumSequential + " in " + duration + "ms");// 并行流start = System.currentTimeMillis();int sumParallel = numbers.parallelStream().mapToInt(ParallelStreams::slowSquare) // 模拟耗时操作.sum();duration = System.currentTimeMillis() - start;System.out.println("Parallel sum: " + sumParallel + " in " + duration + "ms");// 自定义ForkJoinPoolForkJoinPool customPool = new ForkJoinPool(4);start = System.currentTimeMillis();int sumCustomParallel = customPool.submit(() -> numbers.parallelStream().mapToInt(ParallelStreams::slowSquare).sum()).get();duration = System.currentTimeMillis() - start;System.out.println("Custom pool sum: " + sumCustomParallel + " in " + duration + "ms");}// 模拟耗时操作private static int slowSquare(int n) {try {Thread.sleep(100);} catch (InterruptedException e) {e.printStackTrace();}return n * n;}
}
4、原始类型流与flatMap
原始类型流（如IntStream、DoubleStream）避免装箱开销，提供特化方法（如sum、average）。flatMap将嵌套结构（如Stream<List>）扁平化为单一流，例如拆分字符串流为单词流。它先映射后展开，适合处理一对多关系，如stream.flatMap(list -> list.stream())。
import java.util.Arrays;
import java.util.List;
import java.util.stream.IntStream;
import java.util.stream.Stream;public class PrimitiveAndFlatMap {public static void main(String[] args) {// 原始类型流（避免装箱开销）IntStream.rangeClosed(1, 5) // 生成1-5的IntStream.map(n -> n * n) // 平方.average() // 计算平均值.ifPresent(avg -> System.out.println("Average: " + avg));// flatMap示例：将多个流扁平化为一个流List<List<String>> listOfLists = Arrays.asList(Arrays.asList("a", "b"),Arrays.asList("c", "d", "e"),Arrays.asList("f"));// 使用flatMap将嵌套列表扁平化List<String> flatList = listOfLists.stream().flatMap(List::stream) // 将每个List<String>转换为Stream<String>.collect(Collectors.toList());System.out.println("Flattened list: " + flatList);// 更复杂的flatMap示例：生成笛卡尔积List<String> colors = Arrays.asList("Red", "Green", "Blue");List<String> sizes = Arrays.asList("S", "M", "L", "XL");List<String> combinations = colors.stream().flatMap(color -> sizes.stream().map(size -> color + "-" + size)).collect(Collectors.toList());System.out.println("Color-Size combinations: " + combinations);}
}
5、自定义收集器
通过实现Collector<T,A,R>接口可自定义收集逻辑，需定义以下内容，例如实现高效字符串拼接或复杂统计。自定义收集器需确保线程安全（若并行）和正确性。
import java.util.*;
import java.util.function.*;
import java.util.stream.Collector;public class CustomCollector {public static void main(String[] args) {List<String> words = Arrays.asList("hello", "world", "java", "stream", "collector");// 使用自定义收集器连接字符串，并添加前缀和后缀String concatenated = words.stream().collect(new StringConcatenator(", ", "[", "]"));System.out.println("Concatenated: " + concatenated);// 更复杂的自定义收集器：收集统计信息Stats stats = words.stream().collect(Collector.of(Stats::new, // 供应者Stats::accumulate, // 累加器Stats::combine, // 组合器（用于并行流）Stats::finalize // 完成器));System.out.println("Stats: " + stats);}
}// 自定义字符串连接收集器
class StringConcatenator implements Collector<String, StringBuilder, String> {private final String delimiter;private final String prefix;private final String suffix;public StringConcatenator(String delimiter, String prefix, String suffix) {this.delimiter = delimiter;this.prefix = prefix;this.suffix = suffix;}@Overridepublic Supplier<StringBuilder> supplier() {return StringBuilder::new;}@Overridepublic BiConsumer<StringBuilder, String> accumulator() {return (sb, str) -> {if (sb.length() > 0) {sb.append(delimiter);}sb.append(str);};}@Overridepublic BinaryOperator<StringBuilder> combiner() {return (sb1, sb2) -> {if (sb1.length() > 0 && sb2.length() > 0) {sb1.append(delimiter);}return sb1.append(sb2);};}@Overridepublic Function<StringBuilder, String> finisher() {return sb -> prefix + sb.toString() + suffix;}@Overridepublic Set<Characteristics> characteristics() {return Collections.emptySet();}
}// 用于收集统计信息的类
class Stats {private int count;private int totalLength;private String shortest;private String longest;public Stats() {shortest = "";longest = "";}public void accumulate(String word) {if (count == 0) {shortest = longest = word;} else {if (word.length() < shortest.length()) shortest = word;if (word.length() > longest.length()) longest = word;}count++;totalLength += word.length();}public Stats combine(Stats other) {if (this.count == 0) return other;if (other.count == 0) return this;this.count += other.count;this.totalLength += other.totalLength;this.shortest = this.shortest.length() <= other.shortest.length() ? this.shortest : other.shortest;this.longest = this.longest.length() >= other.longest.length() ? this.longest : other.longest;return this;}public Stats finalize() {return this;}@Overridepublic String toString() {return String.format("count=%d, avgLen=%.2f, shortest='%s', longest='%s'",count, (double)totalLength/count, shortest, longest);}
}
Java Stream的中间操作（如filter、map）延迟执行，返回新流；终端操作（如collect、forEach）触发计算，产生结果。高级收集器（Collectors.groupingBy等）支持复杂聚合。并行流利用多核提升性能，但需注意线程安全和开销平衡。原始类型流（IntStream等）避免装箱开销；flatMap合并嵌套流为单一流。自定义收集器通过实现Collector接口满足特定需求，灵活但需手动处理并发。
