Java 8 Stream API 高级实战：从数据处理到性能优化的深度解析
Java 8 引入的 Stream API 彻底改变了 Java 开发者处理集合数据的方式，通过 “声明式编程”“惰性求值”“并行处理” 等特性，大幅简化了数据过滤、映射、聚合等操作。但在实际开发中，多数开发者仅使用filter()、map()、collect()等基础 API，对 Stream 的高级特性（如自定义收集器、并行流优化、状态 ful 中间操作）缺乏深入理解，导致无法充分发挥 Stream 的性能优势，甚至引发线程安全问题。本文将从基础 Stream API 的局限出发，详解 Stream 的高级操作、实战场景、性能优化技巧及避坑指南，帮你从 “会用 Stream” 进阶到 “用好 Stream”。
一、基础 Stream API 的局限：为什么需要高级应用？
在理解 Stream 高级特性之前，我们首先要明确：基础 Stream API 在复杂业务场景（如自定义聚合、并行流性能调优、状态 ful 操作）中存在明显短板，这些局限会导致代码冗余、性能低下或逻辑错误。
1.1 局限 1：基础收集器无法满足自定义聚合需求
Stream 提供的Collectors工具类包含toList()、groupingBy()、summingInt()等基础收集器，但在复杂聚合场景（如 “按用户分组后计算每个用户的订单平均金额与最大金额”“将集合转换为自定义数据结构”）中，基础收集器无法满足需求，需手动编写冗余代码。
示例：基础 API 实现自定义聚合（冗余）
import java.util.*;
import java.util.stream.Collectors;
// 需求：按用户ID分组，计算每个用户的订单总数、总金额、平均金额
public class StreamBasicAggregationDemo {
public static void main(String[] args) {
List<Order> orders = Arrays.asList(
new Order(1L, 101L, new BigDecimal("99.00")),
new Order(2L, 101L, new BigDecimal("199.00")),
new Order(3L, 102L, new BigDecimal("299.00")),
new Order(4L, 102L, new BigDecimal("399.00")),
new Order(5L, 102L, new BigDecimal("499.00"))
);
// 基础API实现：先分组，再遍历计算（冗余）
Map<Long, List<Order>> orderGroupByUserId = orders.stream()
.collect(Collectors.groupingBy(Order::getUserId));
// 手动遍历分组结果，计算聚合指标
Map<Long, UserOrderStats> userStatsMap = new HashMap<>();
for (Map.Entry<Long, List<Order>> entry : orderGroupByUserId.entrySet()) {
Long userId = entry.getKey();
List<Order> userOrders = entry.getValue();
// 计算订单总数
long orderCount = userOrders.size();
// 计算总金额
BigDecimal totalAmount = userOrders.stream()
.map(Order::getAmount)
.reduce(BigDecimal.ZERO, BigDecimal::add);
// 计算平均金额
BigDecimal avgAmount = totalAmount.divide(BigDecimal.valueOf(orderCount), 2, BigDecimal.ROUND_HALF_UP);
userStatsMap.put(userId, new UserOrderStats(orderCount, totalAmount, avgAmount));
}
// 输出结果
userStatsMap.forEach((userId, stats) ->
System.out.printf("用户ID：%d，订单数：%d，总金额：%.2f，平均金额：%.2f%n",
userId, stats.getOrderCount(), stats.getTotalAmount(), stats.getAvgAmount())
);
}
// 订单类
static class Order {
private Long id;
private Long userId;
private BigDecimal amount;
// 构造器、getter
public Order(Long id, Long userId, BigDecimal amount) {
this.id = id;
this.userId = userId;
this.amount = amount;
}
public Long getUserId() { return userId; }
public BigDecimal getAmount() { return amount; }
}
// 用户订单统计类
static class UserOrderStats {
private long orderCount;
private BigDecimal totalAmount;
private BigDecimal avgAmount;
// 构造器、getter
public UserOrderStats(long orderCount, BigDecimal totalAmount, BigDecimal avgAmount) {
this.orderCount = orderCount;
this.totalAmount = totalAmount;
this.avgAmount = avgAmount;
}
public long getOrderCount() { return orderCount; }
public BigDecimal getTotalAmount() { return totalAmount; }
public BigDecimal getAvgAmount() { return avgAmount; }
}
}
问题分析：
- 代码冗余：需先分组再手动遍历，聚合逻辑分散，可读性差；
- 效率低下：分组后需再次遍历流计算指标，存在重复迭代；
- 可维护性差：若新增聚合指标（如最小金额），需修改遍历逻辑，违反 “开闭原则”。
1.2 局限 2：并行流性能不可控（默认线程池与负载均衡问题）
Stream 的parallelStream()提供了简单的并行处理能力，但默认使用ForkJoinPool.commonPool()（全局线程池），存在两大问题：
- 线程资源竞争：全局线程池被所有并行流共享，高并发场景下会导致线程资源竞争，性能下降；
- 负载不均衡：默认的任务拆分策略（按 CPU 核心数拆分）无法适配 “数据倾斜” 场景（如部分数据处理耗时远高于其他数据），导致并行效率低下。
示例：并行流处理数据倾斜场景（性能低下）
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
// 需求：并行处理订单列表，计算每个订单的税额（部分订单需复杂计算，数据倾斜）
public class StreamParallelDataSkewDemo {
public static void main(String[] args) {
List<Order> orders = generateOrders(); // 生成1000个订单，其中10个需复杂计算
long start = System.currentTimeMillis();
// 并行流处理：默认ForkJoinPool，数据倾斜导致性能低下
List<BigDecimal> taxList = orders.parallelStream()
.map(order -> calculateTax(order)) // 部分订单计算耗时100ms，其余1ms
.collect(Collectors.toList());
long end = System.currentTimeMillis();
System.out.printf("并行流处理耗时：%dms，税额列表大小：%d%n", end - start, taxList.size());
}
// 生成订单列表：1000个订单，其中10个标记为“复杂订单”
private static List<Order> generateOrders() {
List<Order> orders = new ArrayList<>();
for (int i = 0; i < 1000; i++) {
boolean isComplex = i < 10; // 前10个为复杂订单
orders.add(new Order((long) i, new BigDecimal(i * 100), isComplex));
}
return orders;
}
// 计算税额：复杂订单耗时100ms，普通订单耗时1ms
private static BigDecimal calculateTax(Order order) {
try {
if (order.isComplex()) {
Thread.sleep(100); // 复杂订单模拟耗时
} else {
Thread.sleep(1); // 普通订单模拟耗时
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return order.getAmount().multiply(new BigDecimal("0.1")); // 税率10%
}
// 订单类
static class Order {
private Long id;
private BigDecimal amount;
private boolean isComplex;
// 构造器、getter
public Order(Long id, BigDecimal amount, boolean isComplex) {
this.id = id;
this.amount = amount;
this.isComplex = isComplex;
}
public BigDecimal getAmount() { return amount; }
public boolean isComplex() { return isComplex; }
}
}
运行结果（CPU 核心数为 8）：
并行流处理耗时：~1200ms，税额列表大小：1000
问题分析：
- 负载不均衡：10 个复杂订单（共耗时 1000ms）被分配到少数线程，其余 990 个普通订单（共耗时 990ms）被分配到其他线程，导致整体耗时由最慢的线程决定（~1200ms）；
- 无自定义线程池：默认线程池无法根据业务场景调整线程数，无法优化数据倾斜问题。
1.3 局限 3：状态 ful 中间操作的线程安全风险
Stream 的中间操作分为 “无状态（stateless）” 和 “有状态（stateful）” 两类：
- 无状态操作：如filter()、map()，处理当前元素无需依赖其他元素；
- 有状态操作：如distinct()、sorted()、limit()，处理当前元素需依赖其他元素的状态（如distinct()需存储已处理的元素）。
在并行流中使用有状态中间操作时，若未注意线程安全，可能导致结果错误或ConcurrentModificationException。
示例：并行流中使用自定义有状态操作（线程不安全）
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;
// 需求：并行流中过滤重复的用户ID（自定义有状态操作，线程不安全）
public class StreamStatefulUnsafeDemo {
public static void main(String[] args) {
// 模拟包含重复用户ID的列表
List<Long> userIds = Arrays.asList(101L, 102L, 101L, 103L, 102L, 104L);
// 自定义有状态过滤：存储已处理的用户ID，过滤重复
List<Long> distinctUserIds = userIds.parallelStream()
.filter(new CustomDistinctFilter())
.collect(Collectors.toList());
System.out.println("去重后的用户ID：" + distinctUserIds);
}
// 自定义有状态过滤器（线程不安全）
static class CustomDistinctFilter implements Predicate<Long> {
// 存储已处理的用户ID（ArrayList非线程安全）
private final List<Long> seenIds = new ArrayList<>();
@Override
public boolean test(Long userId) {
// 并行流中多个线程同时修改seenIds，导致ConcurrentModificationException或重复元素
if (seenIds.contains(userId)) {
return false;
} else {
seenIds.add(userId);
return true;
}
}
}
}
可能的运行结果（错误）：
Exception in thread "main" java.util.ConcurrentModificationException
// 或 去重后的用户ID：[101, 102, 101, 103, 104]（包含重复）
问题分析：
- 线程不安全集合：ArrayList非线程安全，并行流中多个线程同时调用add()和contains()，导致ConcurrentModificationException；
- 状态共享：CustomDistinctFilter的seenIds被所有线程共享，无法保证状态一致性，导致去重结果错误。
1.2 Stream 高级特性的核心价值
Stream 的高级特性通过以下 3 点设计，从根本上解决了基础 API 的局限：
- 自定义收集器：通过Collector接口实现自定义聚合逻辑，支持复杂指标计算，代码简洁且可复用；
- 并行流优化：支持指定自定义ForkJoinPool、自定义任务拆分策略，解决线程资源竞争与负载均衡问题；
- 安全的状态 ful 操作：提供线程安全的有状态中间操作（如distinct()的并行实现），或通过ThreadLocal隔离线程状态，避免线程安全风险。
二、Stream 高级特性：自定义收集器与并行流优化
Stream 的高级特性围绕 “自定义数据处理逻辑” 和 “性能优化” 展开，核心包括自定义收集器、并行流高级配置、状态 ful 操作安全实践三大模块。
2.1 1. 自定义收集器：实现复杂聚合逻辑
Collector接口是 Stream 收集操作的核心，通过实现该接口，可自定义聚合逻辑，支持 “Supplier（初始化容器）→ Accumulator（累加元素）→ Combiner（合并容器，并行场景）→ Finisher（最终处理）” 的完整流程。
1.1 Collector接口核心方法
| 方法 | 作用 | 示例（用户订单统计） |
| supplier() | 提供初始的结果容器 | 返回new UserOrderStats(0, BigDecimal.ZERO, BigDecimal.ZERO) |
| accumulator() | 累加元素到结果容器 | 将订单金额累加到totalAmount，订单数 + 1 |
| combiner() | 合并两个结果容器（并行流） | 合并两个UserOrderStats的orderCount、totalAmount |
| finisher() | 对结果容器进行最终处理 | 计算avgAmount = totalAmount / orderCount |
| characteristics() | 收集器特性（如是否并发、是否无序） | 返回Collections.singleton(Collector.Characteristics.CONCURRENT) |
1.2 示例：自定义收集器实现用户订单统计
import java.util.*;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collector;
import java.util.stream.Collectors;
// 自定义收集器：按用户ID分组，计算每个用户的订单统计信息
public class UserOrderStatsCollector implements Collector<Order, UserOrderStats, UserOrderStats> {
// 1. 初始化结果容器（每个线程一个容器，并行场景）
@Override
public Supplier<UserOrderStats> supplier() {
return () -> new UserOrderStats(0, BigDecimal.ZERO, BigDecimal.ZERO);
}
// 2. 累加元素到容器（将订单金额累加到totalAmount，订单数+1）
@Override
public BiConsumer<UserOrderStats, Order> accumulator() {
return (stats, order) -> {
stats.setOrderCount(stats.getOrderCount() + 1);
stats.setTotalAmount(stats.getTotalAmount().add(order.getAmount()));
};
}
// 3. 合并两个容器（并行流中合并不同线程的统计结果）
@Override
public BinaryOperator<UserOrderStats> combiner() {
return (stats1, stats2) -> {
long combinedCount = stats1.getOrderCount() + stats2.getOrderCount();
BigDecimal combinedTotal = stats1.getTotalAmount().add(stats2.getTotalAmount());
return new UserOrderStats(combinedCount, combinedTotal, BigDecimal.ZERO); // 平均金额暂不计算
};
}
// 4. 最终处理（计算平均金额）
@Override
public Function<UserOrderStats, UserOrderStats> finisher() {
return stats -> {
if (stats.getOrderCount() == 0) {
return stats;
}
// 计算平均金额（保留2位小数）
BigDecimal avgAmount = stats.getTotalAmount()
.divide(BigDecimal.valueOf(stats.getOrderCount()), 2, BigDecimal.ROUND_HALF_UP);
stats.setAvgAmount(avgAmount);
return stats;
};
}
// 5. 收集器特性：CONCURRENT（支持并发累加）、UNORDERED（结果无序）
@Override
public Set<Characteristics> characteristics() {
return Collections.unmodifiableSet(EnumSet.of(
Collector.Characteristics.CONCURRENT,
Collector.Characteristics.UNORDERED
));
}
// -------------- 使用自定义收集器 --------------
public static void main(String[] args) {
List<Order> orders = Arrays.asList(
new Order(1L, 101L, new BigDecimal("99.00")),
new Order(2L, 101L, new BigDecimal("199.00")),
new Order(3L, 102L, new BigDecimal("299.00")),
new Order(4L, 102L, new BigDecimal("399.00")),
new Order(5L, 102L, new BigDecimal("499.00"))
);
// 按用户ID分组，使用自定义收集器计算统计信息
Map<Long, UserOrderStats> userStatsMap = orders.stream()
.collect(Collectors.groupingBy(
Order::getUserId, // 分组键：用户ID
new UserOrderStatsCollector() // 自定义收集器
));
// 输出结果
userStatsMap.forEach((userId, stats) ->
System.out.printf("用户ID：%d，订单数：%d，总金额：%.2f，平均金额：%.2f%n",
userId, stats.getOrderCount(), stats.getTotalAmount(), stats.getAvgAmount())
);
}
// 订单类（同上）
static class Order { /* 省略构造器、getter */ }
// 用户订单统计类（支持setter，用于累加）
static class UserOrderStats {
private long orderCount;
private BigDecimal totalAmount;
private BigDecimal avgAmount;
// 构造器、getter、setter
public UserOrderStats(long orderCount, BigDecimal totalAmount, BigDecimal avgAmount) {
this.orderCount = orderCount;
this.totalAmount = totalAmount;
this.avgAmount = avgAmount;
}
// 省略getter、setter
}
}
关键优势：
- 代码简洁：聚合逻辑集中在收集器中，无需手动遍历分组结果；
- 并行友好：combiner()方法支持并行流的容器合并，无需额外处理线程安全；
- 可复用：自定义收集器可作为工具类复用，减少重复代码。
2.2 2. 并行流优化：自定义线程池与任务拆分
并行流的性能优化核心在于 “线程池隔离” 和 “任务拆分策略优化”，通过指定自定义ForkJoinPool和Spliterator（任务拆分器），可解决默认并行流的性能问题。
2.1 自定义线程池：避免全局线程池竞争
默认并行流使用ForkJoinPool.commonPool()，生产环境中建议使用自定义ForkJoinPool，通过submit()方法提交并行流任务，实现线程池隔离。
示例：自定义线程池优化并行流
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.Collectors;
// 需求：使用自定义线程池处理数据倾斜的订单税额计算
public class StreamParallelCustomPoolDemo {
// 自定义线程池：核心线程数=10（根据CPU核心数和业务场景调整）
private static final ForkJoinPool customForkJoinPool = new ForkJoinPool(10);
public static void main(String[] args) {
List<Order> orders = generateOrders(); // 生成1000个订单（10个复杂订单）
long start = System.currentTimeMillis();
// 使用自定义线程池提交并行流任务
List<BigDecimal> taxList = customForkJoinPool.submit(() ->
orders.parallelStream()
.map(StreamParallelCustomPoolDemo::calculateTax)
.collect(Collectors.toList())
).join(); // 等待任务完成
long end = System.currentTimeMillis();
System.out.printf("自定义线程池处理耗时：%dms，税额列表大小：%d%n", end - start, taxList.size());
// 关闭线程池（生产环境中需在应用关闭时关闭）
customForkJoinPool.shutdown();
}
// 生成订单列表（同上）
private static List<Order> generateOrders() { /* 省略实现 */ }
// 计算税额（同上）
private static BigDecimal calculateTax(Order order) { /* 省略实现 */ }
// 订单类（同上）
static class Order { /* 省略实现 */ }
}
运行结果（自定义线程池核心数 = 10）：
自定义线程池处理耗时：~200ms，税额列表大小：1000
优化原理：
- 线程池隔离：自定义线程池避免与其他并行流共享资源，减少竞争；
- 线程数适配：核心线程数 = 10，可同时处理 10 个复杂订单（每个耗时 100ms），整体耗时降至～200ms（10 个复杂订单分 2 批处理，每批 5 个，共 200ms）。
2.2 自定义 Spliterator：优化任务拆分
Spliterator是 Stream 并行处理的 “任务拆分器”，负责将流拆分为多个子任务，分配给不同线程执行。默认Spliterator按 “固定大小” 拆分（如 ArrayList 按索引拆分），无法适配数据倾斜场景，通过自定义Spliterator可实现 “按数据复杂度拆分”。
示例：自定义 Spliterator 优化数据倾斜拆分
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
// 需求：自定义Spliterator，将复杂订单与普通订单拆分到不同子任务
public class StreamCustomSpliteratorDemo {
private static final ForkJoinPool customPool = new ForkJoinPool(10);
public static void main(String[] args) {
List<Order> orders = generateOrders();
long start = System.currentTimeMillis();
// 1. 自定义Spliterator，拆分复杂订单与普通订单
Spliterator<Order> customSpliterator = new CustomOrderSpliterator(orders);
// 2. 基于自定义Spliterator创建并行流
Stream<Order> parallelStream = StreamSupport.stream(customSpliterator, true);
// 3. 处理流并计算税额
List<BigDecimal> taxList = customPool.submit(() ->
parallelStream.map(StreamCustomSpliteratorDemo::calculateTax)
.toList()
).join();
long end = System.currentTimeMillis();
System.out.printf("自定义Spliterator处理耗时：%dms，税额列表大小：%d%n", end - start, taxList.size());
customPool.shutdown();
}
// 自定义Spliterator：将复杂订单与普通订单拆分为两个子任务
static class CustomOrderSpliterator extends Spliterators.AbstractSpliterator<Order> {
private final List<Order> orders;
private int currentIndex = 0;
private boolean split = false; // 是否已拆分
public CustomOrderSpliterator(List<Order> orders) {
super(orders.size(), Spliterator.ORDERED | Spliterator.SIZED);
this.orders = orders;
}
@Override
public boolean tryAdvance(Consumer<? super Order> action) {
if (currentIndex < orders.size()) {
action.accept(orders.get(currentIndex++));
return true;
}
return false;
}
@Override
public Spliterator<Order> trySplit() {
// 仅拆分一次：将复杂订单与普通订单分为两个子任务
if (split) {
return null; // 已拆分，返回null
}
// 分离复杂订单与普通订单
List<Order> complexOrders = new ArrayList<>();
List<Order> normalOrders = new ArrayList<>();
for (Order order : orders) {
if (order.isComplex()) {
complexOrders.add(order);
} else {
normalOrders.add(order);
}
}
// 更新当前Spliterator为普通订单，返回复杂订单的Spliterator
this.orders.clear();
this.orders.addAll(normalOrders);
this.currentIndex = 0;
this.split = true;
return new CustomOrderSpliterator(complexOrders);
}
}
// 生成订单、计算税额、订单类（同上）
private static List<Order> generateOrders() { /* 省略 */ }
private static BigDecimal calculateTax(Order order) { /* 省略 */ }
static class Order { /* 省略 */ }
}
运行结果：
自定义Spliterator处理耗时：~110ms，税额列表大小：1000
优化原理：
- 按复杂度拆分：将 10 个复杂订单拆分为一个子任务（耗时 100ms），990 个普通订单拆分为另一个子任务（耗时～100ms），两个子任务并行执行，总耗时降至～110ms；
- 动态拆分：trySplit()方法仅拆分一次，避免过度拆分导致的性能开销。
2.3 3. 状态 ful 中间操作的安全实践
在并行流中使用有状态中间操作时，需遵循 “线程隔离” 或 “使用线程安全容器” 的原则，避免线程安全风险。
3.1 安全使用distinct()与sorted()
Stream 的内置有状态操作（如distinct()、sorted()）已实现线程安全，无需额外处理，但需注意：
- distinct()在并行流中使用ConcurrentHashMap存储已处理元素，确保线程安全；
- sorted()在并行流中先局部排序，再合并排序结果（类似归并排序），无需手动同步。
示例：并行流安全使用distinct()
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
// 并行流安全使用distinct()
public class StreamStatefulSafeDemo {
public static void main(String[] args) {
List<Long> userIds = Arrays.asList(101L, 102L, 101L, 103L, 102L, 104L);
// 并行流安全去重（内置distinct()线程安全）
List<Long> distinctUserIds = userIds.parallelStream()
.distinct()
.collect(Collectors.toList());
System.out.println("去重后的用户ID：" + distinctUserIds); // 输出：[101, 102, 103, 104]（顺序可能不同）
}
}
3.2 自定义有状态操作：使用ThreadLocal隔离线程状态
若需自定义有状态操作（如 “统计每个线程处理的元素数量”），可通过ThreadLocal存储线程私有状态，避免线程安全问题。
示例：自定义有状态操作（ThreadLocal 隔离状态）
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
// 需求：并行流中统计每个线程处理的订单数量（自定义有状态操作）
public class StreamStatefulThreadLocalDemo {
// ThreadLocal存储每个线程处理的订单数量
private static final ThreadLocal<AtomicInteger> threadOrderCount = ThreadLocal.withInitial(AtomicInteger::new);
public static void main(String[] args) {
List<Order> orders = Arrays.asList(
new Order(1L), new Order(2L), new Order(3L), new Order(4L),
new Order(5L), new Order(6L), new Order(7L), new Order(8L)
);
// 并行流处理订单，统计每个线程处理的数量
orders.parallelStream()
.forEach(order -> {
// 从ThreadLocal获取当前线程的计数器，累加
AtomicInteger count = threadOrderCount.get();
count.incrementAndGet();
System.out.printf("线程：%s，处理订单ID：%d，当前线程处理总数：%d%n",
Thread.currentThread().getName(), order.getId(), count.get());
});
// 清理ThreadLocal（避免内存泄漏）
threadOrderCount.remove();
}
// 订单类
static class Order {
private Long id;
public Order(Long id) { this.id = id; }
public Long getId() { return id; }
}
}
运行结果（示例）：
线程：ForkJoinPool.commonPool-worker-1，处理订单ID：1，当前线程处理总数：1
线程：ForkJoinPool.commonPool-worker-2，处理订单ID：2，当前线程处理总数：1
线程：ForkJoinPool.commonPool-worker-1，处理订单ID：3，当前线程处理总数：2
线程：ForkJoinPool.commonPool-worker-2，处理订单ID：4，当前线程处理总数：2
// ... 其他线程输出
关键优势：
- 线程隔离：ThreadLocal确保每个线程的计数器独立，避免线程安全问题；
- 无锁性能：无需使用synchronized或AtomicInteger（仅线程内使用），性能高效。
三、Stream 高级实战场景：从业务需求到代码落地
Stream 的高级特性在复杂业务场景中应用广泛，本节将结合 3 个典型场景（复杂报表生成、大数据量并行处理、自定义数据转换），展示从需求分析到高级 Stream API 实现的完整过程。
3.1 场景 1：电商订单复杂报表生成（自定义收集器）
需求：生成电商订单报表，包含以下指标：
- 按支付方式分组（支付宝、微信支付）；
- 每组统计：订单总数、总金额、平均金额、最高金额订单 ID；
- 过滤掉金额小于 10 元的订单。
实现方案：自定义收集器 + 分组聚合
import java.math.BigDecimal;
import java.util.*;
import java.util.function.*;
import java.util.stream.Collector;
import java.util.stream.Collectors;
// 1. 自定义收集器：统计支付方式的订单指标
class PaymentOrderStatsCollector implements Collector<Order, PaymentOrderStats, PaymentOrderStats> {
@Override
public Supplier<PaymentOrderStats> supplier() {
return () -> new PaymentOrderStats(0, BigDecimal.ZERO, BigDecimal.ZERO, null);
}
@Override
public BiConsumer<PaymentOrderStats, Order> accumulator() {
return (stats, order) -> {
// 累加订单数和总金额
stats.setOrderCount(stats.getOrderCount() + 1);
stats.setTotalAmount(stats.getTotalAmount().add(order.getAmount()));
// 更新最高金额订单ID
if (stats.getMaxAmountOrderId() == null ||
order.getAmount().compareTo(stats.getMaxAmount()) > 0) {
stats.setMaxAmount(order.getAmount());
stats.setMaxAmountOrderId(order.getId());
}
};
}
@Override
public BinaryOperator<PaymentOrderStats> combiner() {
return (stats1, stats2) -> {
// 合并订单数和总金额
long combinedCount = stats1.getOrderCount() + stats2.getOrderCount();
BigDecimal combinedTotal = stats1.getTotalAmount().add(stats2.getTotalAmount());
// 合并最高金额订单
PaymentOrderStats maxStats = stats1.getMaxAmount().compareTo(stats2.getMaxAmount()) > 0 ? stats1 : stats2;
return new PaymentOrderStats(
combinedCount,
combinedTotal,
maxStats.getMaxAmount(),
maxStats.getMaxAmountOrderId()
);
};
}
@Override
public Function<PaymentOrderStats, PaymentOrderStats> finisher() {
return stats -> {
// 计算平均金额
if (stats.getOrderCount() > 0) {
BigDecimal avgAmount = stats.getTotalAmount()
.divide(BigDecimal.valueOf(stats.getOrderCount()), 2, BigDecimal.ROUND_HALF_UP);
stats.setAvgAmount(avgAmount);
}
return stats;
};
}
@Override
public Set<Characteristics> characteristics() {
return Collections.unmodifiableSet(EnumSet.of(
Collector.Characteristics.CONCURRENT,
Collector.Characteristics.UNORDERED
));
}
}
// 2. 报表生成主类
public class EcommerceOrderReportDemo {
public static void main(String[] args) {
// 模拟订单数据
List<Order> orders = Arrays.asList(
new Order(1L, "ALIPAY", new BigDecimal("99.00")),
new Order(2L, "ALIPAY", new BigDecimal("199.00")),
new Order(3L, "WECHAT", new BigDecimal("299.00")),
new Order(4L, "WECHAT", new BigDecimal("5.00")), // 金额<10元，会被过滤
new Order(5L, "ALIPAY", new BigDecimal("399.00")),
new Order(6L, "WECHAT", new BigDecimal("499.00"))
);
// 生成报表：过滤→分组→自定义聚合
Map<String, PaymentOrderStats> report = orders.stream()
.filter(order -> order.getAmount().compareTo(new BigDecimal("10.00")) >= 0) // 过滤小额订单
.collect(Collectors.groupingBy(
Order::getPaymentMethod, // 按支付方式分组
new PaymentOrderStatsCollector() // 自定义收集器
));
// 输出报表
System.out.println("电商订单支付方式报表：");
report.forEach((paymentMethod, stats) -> {
System.out.printf("支付方式：%s%n", paymentMethod);
System.out.printf(" - 订单总数：%d%n", stats.getOrderCount());
System.out.printf(" - </doubaocanvas>