JavaPriorityQueue多源日志排序实现

public static class LogEntry implements Comparable {
    public final long timestamp;
    public final String source;
    public final String content;
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); // 升序：时间早的排前面
}
}

public static class LogStreamCursor {
    public final Iterator iterator;
    public final String sourceName;
    public LogEntry next; // 当前已拉取、待参与比较的条目
public LogStreamCursor(Iterator<LogEntry> it, String name) {
    this.iterator = it;
    this.sourceName = name;
    advance(); // 预读第一条
}

public void advance() {
    this.next = iterator.hasNext() ? iterator.next() : null;
}
}

public static Stream mergeSortedLogs(List> streams) {
    PriorityQueue queue = new PriorityQueue<>(
        Comparator.comparing(cursor -> cursor.next.timestamp)
    );
// 初始化：每路游标预读成功后入队
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);
}