您的位置: 首页  >  文章  >  Java 8 Stream API 实用指南

Java 8 Stream API 实用指南

分类: 文章 • 2025-03-11 07:43:45

Java 8 Stream API 实用指南

本文来自作者 阿福 在 GitChat 上分享「Java 8 Stream API 实用指南」,「阅读原文」查看交流实录

「文末高能」

编辑 | 一雄

本文作为 Stream API 的 “使用指南”，主要侧重于 “实用”，并不会关注太多的实现细节，当然，不是简单地罗列接口，而是尽可能地向读者展示 Stream API 的全貌。

开始之前

作为 Java API 的新成员，Stream API “允许以声明式的方式处理数据集合”。回顾 “内容介绍” 部分，我们阅读了以下的代码：



class Good {
 String name;               // 商品名称
 long price;               // 价格
 long sales;               // 销量
 Listcategories;  // 类别
 // ... 省略 constructor、getter / setter
 // ... 省略 toString
}
void process(Listgoods) {
 //
 // 筛选 price > 500 & sales < 200 的商品, 价格最高的 10 件商品, 价格减半（双十一来啦！）
 //
 goods.stream()
   .filter(c -> c.getPrice() > 500 && c.getSales() < 200)
   .sorted(Comparator.comparing(Good::getPrice).reversed())
   .limit(10)
   .forEach(c -> { c.setPrice(c.getPrice() / 2); });
}



即使没有 Stream API，我们依然能够通过完成需求，但无法做到如此简洁、清晰。（本文最后的部分，还会进一步探讨 Stream API 在并发方面的优势。）



开始使用 Stream API 之前，我们需要了解，Stream 是什么？有哪些比较重要的概念？为此，我们针对上文的代码绘制了示意图：






Java 8 Stream API 实用指南



图中所示，整个过程就是将 goods 元素集合作为一个 “序列”，进行一组 “流水线” 操作，其中：


    

  1. 

    goods 集合提供了元素序列的数据源，通过 stream() 方法获得 Stream
    

    2. 

  2. 

    filter / sorted / limit 进行数据处理，“连接起来” 构成 “流水线”
    

    3. 

  3. 

    forEach 最终执行
    

    4. 




需要说明，filter / sorted / limit 的返回值均为 Stream（类似于 Builder 模式），但它们并不立即执行，而是构成了 “流水线”，直到 forEach：最终执行，并且关闭 Stream。因此：


    

  1. 

    将 filter / sorted / limited 等能够 “连接起来”，并且返回 Stream 的方法称为 “中间操作”（Intermediate）
    

    2. 

  2. 

    将 forEach 等最终执行，并且关闭 Stream 的方法称为 “终止操作” （Terminal）
    

    3. 




特别地，需要记住：Stream 的中间操作并不是立即执行，而是 “延迟的”、“按需计算”；并且，完成 “终止操作” 后，Stream 将被关闭。



现在，我们应当了解 Stream 的关键概念：数据源、中间操作构成 “流水线”、终止操作，对于 Stream 的定义，我们直接引用 Java doc：






A sequence of elements supporting sequential and parallel aggregate operations.




package java.util.stream;
public interface Streamextends BaseStream<t, stream



代码所示：Stream 中序列元素的类型，通过泛型表达。对于原始类似，除了包装类外，Stream API 同时提供了原始类型的 Stream：IntStream、LongStream 以及 DoubleStream



本质而言，Stream API 的 “流水线” 操作，最终仍然依赖于迭代，但与使用 Collection API 直接构建迭代的代码不同，Stream API 通过参数的形式接收我们提供的操作，由其内部实现迭代。



而所谓 “我们提供的操作”，请参考 “Lambda 表达式 & 方法引用”。



Lambda 表达式 & 方法引用




Lambda 表达式，或者 λ，虽然并不是新鲜事物，但其越来越受到重视，尤其是 Java 8（以及 C++11）将 Lambda 表达式纳入标准以后。暂且不谈论 “函数式编程” 的话题，我们先了解 Lambda 表达式如何使用。



Lamba 表达式




对于 Java 开发者，匿名类是很常见的东西，例如：




@FunctionalInterface
interface PriceCalculator {
 long calculate(Good good);
};
public void process(Listgoods, PriceCalculator calculator) {
 // 计算商品价格    
}
//
// 实现 PriceCalculator 接口的匿名类实例，作为 process 参数
//
process(goods, new PriceCalculator() {
 @Override
 public long calculate(Good good) {
   return good.getPrice();
 }
});



接口 PriceCalculator 只有一个方法，我们将只有一个抽象方法的接口，称为 “函数式接口”，并以 @FunctionalInterface 进行标记。（注意，Java 8 允许接口提供方法实现，即 “默认方法”，函数式接口必须包含且仅包含一个抽象方法，对于提供实现的默认方法，没有限制）



Lambda 表达式，其本质即为函数式接口的一个实例：




//
// 示例 #1: args -> { statement; }
//
process(goods, (good) -> {
 return good.getPrice();
});
//
// 示例 #2：args -> expression
//
process(goods, (good) -> good.getPrice());



函数式接口中抽象方法的签名即为 Lambda 表达式的签名，称为 “函数描述符”。Lambda 表达式的类型，由 Java 编译器根据上下文推断获得。



方法引用




方法引用，即为特定情况下 Lambda 表达式的简化，例如：




process(goods, Good::getPrice);



对于 Lambda 表达式到方法引用的简化，我们提供以下规则：






Java 8 Stream API 实用指南



特别的，对于构造函数的方法引用：ClassName::new



开始使用 Stream API



本章节将阐述 Stream 的生成、操作、数据收集，主要介绍 Stream API 的常用接口与辅助方法。为了便于我们试验示例的代码，我们先说明 forEach(Consumer<? super T>)。



正如前面章节所说，forEach(Consumer<? super T>) 是一个 “终止操作”，它遍历 Stream 的元素序列，通过函数式接口 Consumer<? super T> 的 accept(T) 执行特定操作。Consumer<? super T> 的声明：




@FunctionalInterface
public interface Consumer{
 void accept(T);
}



以下的阐述中，将通过 forEach(System.out::println) 将 Stream 的元素序列输出。



生成 Stream




由集合 & 数组生成 Stream



Stream 作为元素的 “序列”，自然而然地，我们想到通过集合、数组生成 Stream。



Java 8 的 Collection 接口添加了 Streamstream() 方法，由集合生成 Stream，例如：




//
// 输出商品集合
//
void print(Listgoods) {
 goods.stream().forEach(System.out::println);
}



java.util.Arrays 提供了 stream(T[]) 的静态方法，由 T[] 数组生成 Stream：






Java 8 Stream API 实用指南






特别地，当数组元素类型 T 是原始类型，静态方法 stream(T[]) 将返回原始类型的 Stream。



通过集合或数组获得的 Stream，是 “有限” 的。



直接创建 Stream



除了由集合和数组生成 Stream，Stream API 提供了静态方法 Stream.generate(Supplier)、Stream.iterator(final T, final
UnaryOperator)，直接创建 Stream。



Stream.generate(Supplier) 通过参数 Supplier
获取 Stream 序列的新元素




//
// 生成指定数量的商品并输出
//
void generate(int number) {
 Stream.generate(Good::new).limit(number).forEach(System.out::println);
}



Stream.iterator(final T, final UnaryOperator) 提供了一种 “迭代” 的形式：第一个元素，以及第 n 个元素到第 n + 1 个元素的生成方式
UnaryOperator。




//
// 生成指定数量的序列 1, 2, 4, 8, 16 ... 并输出
//
void generateSequence(int number) {
 Stream.iterate(0, n -> n * 2).limit(number).forEach(System.out::println);
}



通过 Stream.generate(Supplier)、Stream.iterator(final T, final UnaryOperator)，将产生
“无限的” Stream，以上的示例中，使用 limit 进行了 Stream 截断。



操作 Stream




filter



filter 是 “中间操作”，以 Predicate<? super T> 的实例作为参数，进行 Stream 过滤，仅保留符合条件的元素。Predicate<? super T> 作为常用的函数式接口，其声明如下：




@FunctionalInterface
public interface Predicate{
   boolean test(T);
}


例如：




//
// 过滤高于指定价格的商品
//
void filterByPrice(Listgoods, long price) {
 goods.stream().filter(c -> c.getPrice() > price).forEach(System.out::println);
}



anyMatch / allMatch / noneMatch




anyMatch、allMatch、noneMatch，都是 “终止操作”，与 filter 接收相同的参数，其功能顾名思义，例如：




//
// 检查商品集合是否包含指定名称的商品
//
boolean hasGoodWithName(Listgoods, String name) {
 return goods.stream().anyMatch(c -> name.equals(c.getName()));
}



findAny / findFirst



findAny、findFirst，都是 “终止操作”，分别获取 Stream 元素序列的任意元素和第一个元素：




//
// 获取商品集合中任意名称为指定名称的商品
//
OptionalfindAnyGoodWithName(Listgoods, String name) {
 return goods.stream().filter(c -> name.equals(c.getName())).findAny();
}



findAny、findFirst 的返回值都是 Optional 类型，避免了 Stream 序列为空时返回 null。关于 Optional 类型，不属于本文的范围，请参阅 Java
doc。



相比较于 findFisrt，findAny 更适合于并发的场景。



map



map 是中间操作，将 Stream 序列的元素映射为其他的元素，以 Function<? super T, ? extends R> 作为参数，其声明如下：




@FunctionalInterface
public interface Function{
   R apply(T);
}



代码所示，Function<? super T, ? extends R> 提供了 Stream 序列的元素映射为其他元素的途径，例如：




//
// 输出商品的名称
//
void printName(Listgoods) {
 goods.stream().map(Good::getName).forEach(System.out::println);
}



此外，Stream 提供 mapToInt、mapToLong、mapToDouble，将 Stream 映射为原始类型 Stream。



flatMap



map 直接将 Stream 序列的元素映射到新的元素，假如 map 映射获得的是 Stream，flatMap 能够将各个 Stream 的元素合并到一个 Stream 中，例如：




//
// 获取商品集合的分类
//
void getCategories(Listgoods) {
 goods.stream().flatMap(c -> c.getCategories().stream()).forEach(System.out::println);
}



distinct



distinct 是 “中间操作”，即去重，去重的依据即为 Stream 序列元素类型的 equals 和 hashCode 方法，例如：




//
// 获取商品名称，去重
//
void distinctGoodNames(Listgoods) {
 goods.stream().map(Good::getName).distinct().forEach(System.out::println);
}



sorted



sorted 是 “中间操作”，以 Comparator<? super T> 作为参数，将 Stream 序列元素排序，Comparator<? super T>：




//
// 商品按照价格升序排列
//
void sortGoods(Listgoods) {
 goods.stream().sorted(Comparator.comparing(Good::getPrice)).forEach(System.out::println);
}



示例代码中，使用辅助方法 Comparatorcomparing(Function<? super T, ? extends U>) 生成了 Comparator<? super T>
实例。



“内容介绍” 部分的 reversed()，同样是 Comparator 的方法，并提供了默认实现，用于排序时，即可实现排序 “取反”。



limit / skip



limit / skip 是 “中间操作”，接收 long 类型的参数，实现 Stream 序列元素的截取和跳过：




//
// 获得第 page 页的商品，每页商品数量为 page_size
//
void listGoods(Listgoods, int page, int page_size) {
 goods.stream().skip((page - 1) * page_size).limit(page).forEach(System.out::println);
}



收集数据




count / min / max



count 是终止操作，将直接返回 Stream 的元素数量：




//
// 获取高于指定价格的商品数量
//
long countGoodsOverPrice(Listgoods, long price) {
 return goods.stream().filter(c -> c.getPrice() > price).count();
}



min / max，以 Comparator<? super T> 作为参数，返回最小值和最大值。对于原始类型 Stream，min / max 无参数，例如：




//
// 获取最高的商品价格
//
OptionalLong maxGoodPrice(Listgoods) {
 return goods.stream().mapToLong(Good::getPrice).max();
}



示例代码中，LongStream 的 max 方法返回类型为 OptionalLong，即为原始类型的 Optional。



reduce



reduce，“归约”，是 “终止操作”，用于将 Stream 序列归约到一个具体的值，其声明，如下：




//
// 提供初始值，以及两个 Stream 序列元素结合产生新值的方法
//
T reduce(T, BinaryOperator);
//
// 提供两个 Stream 序列元素结合产生新值的方法，没有初始值，但通过 Optional避免 Stream 为空时返回 null
//
Optionalreduce(BinaryOperator);
//
// 归约到新的类型：提供初始值，新值与 Stream 元素结合的方法，以及两个新值结合的方法
// U reduce(U, BiFunction, BinaryOperator);




例如：




long getTotalSalesAmount(Listgoods) {
 //
 // 获取 goods 集合的销售总额
 //
 return goods.stream().reduce(0L, (amount, good) -> amount + good.getSales() * good.getPrice(),(left, right) -> left + right);
}


或者：




long getTotalSalesAmount(Listgoods) {
 //
 // 获取 goods 集合的销售总额
 //
 return goods.stream().mapToLong(c -> c.getPrice() * c.getSales()).reduce(0, Long::sum);
}



collect



作为 “终止操作”，collect 即 “收集数据”。collect 以 “收集器” Collector<? super T, A, R> 作为参数，通常，我们使用 Collectors 提供的辅助函数获得 “收集器” 实例。




常用的辅助函数


toList / toSet


 


// // 获取商品名称的集合 // ListgetGoodNames(Listgoods) {   return
goods.stream().map(Good::getName).collect(Collectors.toList()); }


toSet 与 toList 相似，但其返回结果为 Set。


groupingBy


 


// // 将商品集合按照价格分组 // Map<long, listgoods) {   return
goods.stream().collect(Collectors.groupingBy(Good::getPrice)); }


partitioningBy


partitioningBy 与 groupingBy 类似，但其得到分组的键类型为 Boolean，即 true &

false，最多两组：


 


// // 根据是否超过指定销量将商品集合分组 // Map<boolean, listgoods, long sales) {   return
goods.stream().collect(Collectors.partitioningBy(c -> c.getSales() >=
sales)); }


reducing


Collectors.reducing
参数与 
Stream.reduce

一致，其获得的 “收集器” 实例，作为 collect 参数，能够与 reduce 获得相同的结果。





根据是否需要存储中间状态，Stream 操作能够划分为 “无状态操作”、“有状态操作”。“无状态操作”，例如：filter、map；“有状态操作”。



例如：limit、sorted 等，并且，对于 “有状态操作”，亦根据中间状态存储的要求，其区分 “有界”（例如：limit）、“无界”（例如：sorted）。



并行




通过 “并行 Stream” 即可获得 Stream API 的并行能力，例如：




//
// 获取最高的商品价格
//
OptionalLong maxGoodPrice(Listgoods) {
 return goods.stream().parallel().mapToLong(Good::getPrice).max();
}



代码所示，通过 Collection  接口的 parallelStream()、 BaseStream 接口的 parallel() 方法，都能够获得 “并行 Stream”。



并行 Stream 内部是基于 ForkJoinPool 模型获得并行能力，其默认线程数量即为通过 Runtime.getRuntime().availableProcessors() 获得的线程数。



不过，关于并行，两件事必须注意：一方面，正确性，避免 Stream 处理过程中共享可变状态；另一方面，务必记住，并行未必能够提高性能，通常适用于 Stream 元素数量大、或单个元素处理非常耗时的场景。



写在最后




请在阅读完本文后，尝试解答以下问题，最终的答案我们线上交流见 ^_^



1. 以下代码执行，将输出什么？




String[] words = { "a", "bb", "ccc", "dddd", "eeee" };
Arrays.stream(words).filter(c -> { System.out.println(c); return true; }).limit(1).collect(Collectors.toList());


2. 分别运用 Stream API 的 reduce、collect 方法实现以下方法：




long getTotalSalesAmount(Listgoods) {
 //
 // 获取 goods 的销售总金额
 //
}


3. 通过 Stream API 实现以下方法：




void printFibonacciSequence(int n) {
 //
 // 输出斐波那契数列的前 n 个数
 //
}


4. 通过 Stream API 改造以下代码： （提示，需要了解 collect 方法的参数类型 Collector）




class SaleRecord {
   String recordId;  // 销售记录 Id
   int goodId;       // 商品 Id
   int promotionId; // 促销活动 Id
   long price;       // 价格
   long sales;       // 数量
}
class SalesAggregation {
   int goodId;       // 商品 Id
   int promotionId; // 促销活动 Id
   long amount;     // 金额
}
//
// 聚合
//
Listaggregate(ListsaleRecords) {
 MapsalesAggregations = new HashMap<>();
 for (SaleRecord saleRecord : saleRecords) {
   String key = String.format("%d_%d", saleRecord.goodId, saleRecord.promotionId);
   if (salesAggregations.containsKey(key)) {
     salesAggregations.get(key).amount = salesAggregations.get(key).amount + saleRecord.price * saleRecord.sales;
   } else {
     SalesAggregation salesAggregation = new SalesAggregation();
     salesAggregation.goodId = saleRecord.goodId;
     salesAggregation.promotionId = saleRecord.promotionId;
     salesAggregation.amount = saleRecord.price * saleRecord.sales;
     salesAggregations.put(key, salesAggregation);
   }
 }
 return new ArrayList<>(salesAggregations.values());
}



近期热文



《3分钟了解入门「机器学习」该学习什么？（上）》



《3分钟了解入门「机器学习」该学习什么？（下）》



《正则表达式从入门到实战》



《如何用
Node.js 爬虫？》



《两款敏捷工具，治好你碎片化交付硬伤》



福利



Java 8 Stream API 实用指南



「阅读原文」看交流实录，你想知道的都在这里


															

																					

								

									相关推荐