怎么利用 Stream.peek() 在复杂的流处理链路中插入监控点以记录每个阶段的转化耗时-java教程-PHP中文网

Stream.peek() 不改变流元素，仅用于轻量监控（如日志、计时），需终端操作触发；阶段耗时需前后peek配合原子变量记录，推荐封装TimingPeek工具类实现可复用计时，并注意并行流乱序、线程安全及生产环境日志采样。

stream.peek() 本身不改变流元素，但可以在不干扰逻辑的前提下“窥探”每个元素经过时的状态。它适合插入轻量级监控点，比如打日志、计时、采样统计。但要注意：peek() 只在元素被消费时触发，且**不会强制执行流**——必须有终端操作（如 collect、foreach、count）才会真正运行。

用 peek() 搭配 AtomicLong 或 ThreadLocal 实现阶段耗时记录

单个 peek() 无法直接测“阶段耗时”，因为它是按元素逐个调用的。正确做法是：在每个关键处理步骤前后各加一个 peek()，用原子变量记录起止时间，再按元素 ID 或批次聚合耗时。

为每个流元素生成唯一追踪 ID（如 UUID 或递增序号），放在包装对象里，贯穿整个链路
用 AtomicLong startNs 记录进入某阶段的纳秒时间，在第一个 peek() 中存入元素上下文；在下一个 peek() 中读取并计算差值
避免在 peek() 中做阻塞或复杂计算，否则会拖慢整个流，甚至破坏并行流的行为

典型监控链路示例（含耗时记录）

假设你有一条流：从原始字符串 → 解析为对象 → 过滤无效项 → 转换为 DTO → 聚合统计。你想知道每一步对每个元素花了多久：

AtomicLong parseStart = new AtomicLong();
AtomicLong filterStart = new AtomicLong();
AtomicLong dtoStart = new AtomicLong();
<p>List<result> results = lines.stream()
// 阶段1：解析前打点
.peek(line -> parseStart.set(System.nanoTime()))
.map(line -> {
var parsed = parseLine(line); // 实际解析逻辑
long cost = System.nanoTime() - parseStart.get();
log.debug("parse-cost: {} ns for {}", cost, line);
return parsed;
})
// 阶段2：过滤前打点
.peek(obj -> filterStart.set(System.nanoTime()))
.filter(obj -> isValid(obj))
.peek(obj -> {
long cost = System.nanoTime() - filterStart.get();
log.debug("filter-cost: {} ns for {}", cost, obj.id());
})
// 阶段3：转换前打点
.peek(obj -> dtoStart.set(System.nanoTime()))
.map(this::toDto)
.peek(dto -> {
long cost = System.nanoTime() - dtoStart.get();
log.debug("dto-convert-cost: {} ns for {}", cost, dto.id());
})
.collect(Collectors.toList());</result></p>