Java中用PriorityQueue合并多个有序日志流，本质是N路归并：每流用LogStreamCursor预取最小日志入队，每次poll后从对应流advance并重新入队，实现O(N)空间复杂度的惰性归并。
Java 中用 PriorityQueue 合并多个按时间戳有序的日志流，核心是把每个日志流的“当前最小时间戳日志”作为候选元素入队，每次取出全局最小后，再从对应流中补充下一条——这本质上是 N 路归并（N-way merge） 的经典实现。
准备日志数据结构与比较逻辑
定义一个轻量日志类，包含时间戳和来源标识，关键是要让 PriorityQueue 能按时间戳升序排序：
public static class LogEntry implements Comparable<LogEntry> {
public final long timestamp;
public final String source;
public final String content;
<pre class="brush:php;toolbar:false;">public LogEntry(long timestamp, String source, String content) {
this.timestamp = timestamp;
this.source = source;
this.content = content;
}
@Override
public int compareTo(LogEntry other) {
return Long.compare(this.timestamp, other.timestamp); // 升序：时间早的排前面
}}
注意：不要用 timestamp - other.timestamp 防止整数溢出；若需处理毫秒级重复时间戳，可追加 source 或序列号作为第二排序键。
立即学习“Java免费学习笔记（深入）”；
封装每个日志流为可迭代的“游标”
假设每路日志流是 Iterator<logentry></logentry>（如从文件、队列或数据库分页读取），需为每路维护一个“当前未消费的条目”。推荐用简单包装类：
public static class LogStreamCursor {
public final Iterator<LogEntry> iterator;
public final String sourceName;
public LogEntry next; // 当前已拉取、待参与比较的条目
<pre class="brush:php;toolbar:false;">public LogStreamCursor(Iterator<LogEntry> it, String name) {
this.iterator = it;
this.sourceName = name;
advance(); // 预读第一条
}
public void advance() {
this.next = iterator.hasNext() ? iterator.next() : null;
}}
这样每个游标始终持有“本流中下一个可用的最小日志”，避免重复拉取或漏判空流。
构建优先队列并执行归并
初始化时，将每路非空游标的 next 入队；每次 poll 后，从对应游标再 advance() 并重新入队（如果还有下一条）：
public static Stream<LogEntry> mergeSortedLogs(List<Iterator<LogEntry>> streams) {
PriorityQueue<LogStreamCursor> queue = new PriorityQueue<>(
Comparator.comparing(cursor -> cursor.next.timestamp)
);
<pre class="brush:php;toolbar:false;">// 初始化：每路游标预读成功后入队
for (int i = 0; i < streams.size(); i++) {
LogStreamCursor cursor = new LogStreamCursor(streams.get(i), "source-" + i);
if (cursor.next != null) {
queue.offer(cursor);
}
}
return Stream.generate(() -> {
if (queue.isEmpty()) return null;
LogStreamCursor head = queue.poll();
LogEntry result = head.next;
head.advance(); // 拉取本流下一条
if (head.next != null) {
queue.offer(head); // 有后续则放回队列参与下次竞争
}
return result;
}).takeWhile(Objects::nonNull);}
- 使用
Stream.generate+takeWhile实现惰性求值，适合处理海量日志 - 若需阻塞式实时合并（如 Kafka 多分区消费），可改用循环 +
poll+ 条件等待，配合offer新日志 - 注意线程安全：若多线程向不同游标写入新日志，需对
queue和游标状态加锁，或改用ConcurrentSkipListSet替代
边界情况与优化提示
实际部署中需关注：
-
空流或全空流：初始化阶段跳过
next == null的游标，避免队列为空时首次poll返回 null - 时间戳精度问题：若多流来自不同系统，存在时钟漂移，建议统一转换为 UTC 时间戳，并考虑添加“接收时间”辅助字段用于去重或调试
- 内存控制：每路游标只缓存一条日志，空间复杂度为 O(N)，远优于全量加载再排序的 O(M)（M 为总日志数）
-
性能微调：对超大 N（如上百路流），可将
PriorityQueue替换为更高效的堆实现（如 Apache CommonsBinaryHeap），但通常 JDK 默认实现已足够
不复杂但容易忽略。
