分析 Java Stream 的 peek使用实践与副作用处理方案

Stream<T> peek(Consumer<? super T> action);

List<Integer> nums = Arrays.asList(1, 2, 3, 4);
List<Integer> doubled = nums.stream()
    .peek(n -> System.out.println("原始值: " + n))  // 观察元素
    .map(n -> n * 2)
    .peek(n -> System.out.println("翻倍后: " + n))  // 观察转换后的值
    .collect(Collectors.toList());

// 危险示例：并行流中修改共享列表
List<Integer> result = new ArrayList<>();
Arrays.asList(1, 2, 3, 4).parallelStream()
    .peek(n -> result.add(n * 2))  // 多线程同时添加元素
    .collect(Collectors.toList());
// 可能抛出ConcurrentModificationException，或元素重复/丢失

// 错误示例：peek修改元素导致排序混乱
List<User> users = Arrays.asList(
    new User("Alice", 25),
    new User("Bob", 20)
);
users.stream()
    .peek(u -> u.setAge(u.getAge() + 5))  // 修改年龄
    .sorted(Comparator.comparingInt(User::getAge))
    .forEach(u -> System.out.println(u.getName() + " " + u.getAge()));
// 排序依据的是修改后的年龄，但peek的执行顺序可能与排序逻辑冲突

// 低效示例：每行日志都执行peek打印
List<String> logs = Files.readAllLines(path);
long errorCount = logs.stream()
    .peek(System.out::println)  // 每行都打印，IO开销巨大
    .filter(l -> l.contains("ERROR"))
    .count();

// 调试流操作链
List<String> result = dataStream
    .peek(s -> System.out.println("过滤前: " + s))
    .filter(this::validate)
    .peek(s -> System.out.println("过滤后: " + s))
    .map(this::transform)
    .collect(Collectors.toList());

// 安全示例：单线程流中复制对象
List<Product> products = originalList.stream()
    .peek(p -> p = new Product(p))  // 浅拷贝
    .collect(Collectors.toList());

// 仅在收集时执行副作用
AtomicInteger counter = new AtomicInteger();
List<String> result = dataStream
    .peek(s -> {
        if (counter.incrementAndGet() % 1000 == 0) {
            log.info("处理了1000个元素");  // 惰性日志输出
        }
    })
    .collect(Collectors.toList());

// 反例：peek修改对象属性（副作用）
users.stream()
    .peek(u -> u.setStatus("ACTIVE"));  // 直接修改原对象
// 优化：用map创建新对象（无副作用）
List<User> activeUsers = users.stream()
    .map(u -> new User(u.getId(), u.getName(), "ACTIVE"))
    .collect(Collectors.toList());

// 危险：peek修改共享计数器
AtomicInteger count = new AtomicInteger();
dataStream.parallel()
    .peek(s -> count.incrementAndGet());  // 多线程竞争
// 安全：用Collector统计
long safeCount = dataStream.parallel()
    .collect(Collectors.counting());

// 低效：流中执行打印
dataStream.forEach(s -> {
    process(s);
    System.out.println(s);  // 副作用
});
// 高效：先处理再统一输出
List<String> processed = dataStream.map(this::process).collect(Collectors.toList());
processed.forEach(System.out::println);