Salesforce UDP 数字流排序：位移不超过 n 的有界最小堆

题意与假设

这道练习题保留了面经中值得训练的核心约束，并将需要澄清的条件单独列出。现场拿到变体时，先和面试官确认这些条件。

• 训练版把数字本身作为升序排序键，k=max_displacement 表示某元素在输入位置与最终排序秩之间相差不超过 k；若面试官真正指 UDP sequence number、去重或缺包处理，需要另定义协议。
• 允许相同数值；相同键的稳定顺序未要求。输出用生成器逐项交给调用者，不在函数内收集所有结果。
• 输入流有限，k 为非负整数；若流无限，不能最终 flush，但每读入新项可持续发出前面已确定的项。

解题路线

1. 先从流中读入最多 k+1 个数，放进最小堆。随后每次弹出堆顶并输出，再读入一个新数入堆；输入耗尽后依次弹出剩余堆元素。
2. 正确性关键是秩位移保证：全局最小的剩余元素在当前待读窗口最远也只可能比它的正确位置晚 k 个输入位置。因此输出第 r 个元素时，读取到原输入第 r+k 个位置即可确保它已在候选堆中；堆顶就是该轮应输出的最小值。归纳得到完整升序输出。
3. 堆大小始终不超过 k+1，满足不能保存全流的要求；若 k 接近总流长，内存优势也会消失。数据若违反位移界，算法不会发现其未来存在更小值，必须把输入约束作为前提或增加校验缓冲。

手动走一遍

k=2，输入 [3,1,2,5,4]。先装 [3,1,2]，弹 1；读 5，再弹 2；读 4，再弹 3；输入结束后弹 4、5，得到 [1,2,3,4,5]。k=0 表示输入本身已排序，只需一项缓冲逐个输出。

Python 参考实现

下面的代码按上文假设实现，适合本地运行并追加自己的测试用例。

代码区可横向滚动；使用键盘时，先聚焦代码区再按方向键查看。

import heapq
from collections.abc import Iterable, Iterator


def ordered_packets(values: Iterable[int], max_displacement: int) -> Iterator[int]:
    """Yield ascending values when every item is at most k positions from its rank."""
    if max_displacement < 0:
        raise ValueError("max_displacement must be nonnegative")
    source = iter(values)
    heap: list[int] = []
    for _ in range(max_displacement + 1):
        try:
            heapq.heappush(heap, next(source))
        except StopIteration:
            break
    while heap:
        yield heapq.heappop(heap)
        try:
            heapq.heappush(heap, next(source))
        except StopIteration:
            pass

复杂度

m 为流中元素数。每个数入堆、出堆各一次，时间 O(m log(min(k+1,m)))，辅助内存 O(min(k+1,m))；生成器不计算结果列表。若输出接收方自行收集列表，那是调用方额外 O(m) 空间，不计入本函数。

易错点与追问

容易失分的边界

• 仅缓存 k 个元素；位移恰好为 k 的正确下一项可能在第 k+1 个输入位置。
• 先对全流 sort 再输出，违反不可保存所有数字的限制。
• 把网络包排序、丢包重传与数字排序混为一谈；题目没有说明 UDP 协议层的 sequence number、超时和缺包策略。
• 在位移条件不成立的输入上仍承诺全局有序，例如 k=1 却让最小值延迟到流尾。

继续深挖

• 若包会重复、缺失或超过等待期限，需按 sequence number 定义 watermark、去重表和 gap timeout；这已是另一道流处理设计题。
• 若必须稳定排序重复值，堆中保存 (value,inputIndex) 而非仅 value，额外索引空间仍 O(k)。
• 若 k 未知，如何在线调整缓冲？不能在不设延迟上限的同时保证绝对有序，应先协商最大乱序界。