实现内存任务调度器：重复待执行 taskID 忽略，按 runAt 升序 poll 到期任务，数量不超过 limit 且不重复返回。

核心回答

用最小堆保存 (runAt, sequence, taskID, payload)，堆顶始终是最早应执行的任务；用集合记录仍待执行的 taskID，重复提交直接忽略。poll(now, limit) 只弹出到期任务，最多返回 limit 条，并在弹出时同步移除去重集合中的标识。

堆与集合必须由同一把锁保护，否则两个调用者可能重复领取，或发生堆中有任务、集合却已丢失标识的情况。相同时间用递增序号打破并列，既保持提交顺序，也避免比较不可排序的任务对象。

Python 实现

这里 run_at 与 now 使用同一时间基准，单位由调用者统一；task_id 在待执行期间去重，领取之后允许再次提交同名任务。

import heapq
import math
from threading import Lock

class MemoryScheduler:
    def __init__(self):
        self.heap, self.pending = [], set()
        self.sequence = 0
        self.lock = Lock()

    def submit(self, task_id, run_at, payload=None):
        if not isinstance(task_id, str) or not task_id:
            raise ValueError("task_id must be a nonempty string")
        if not math.isfinite(run_at):
            raise ValueError("run_at must be finite")
        with self.lock:
            if task_id in self.pending:
                return False
            heapq.heappush(self.heap, (run_at, self.sequence, task_id, payload))
            self.sequence += 1
            self.pending.add(task_id)
            return True

    def poll(self, now, limit):
        if not math.isfinite(now):
            raise ValueError("now must be finite")
        if type(limit) is not int or limit < 0:
            raise ValueError("limit must be a nonnegative integer")
        result = []
        with self.lock:
            while self.heap and len(result) < limit and self.heap[0][0] <= now:
                at, seq, task_id, payload = heapq.heappop(self.heap)
                self.pending.remove(task_id)
                result.append((task_id, at, payload))
        return result

几个容易写错的边界

• poll() 先判断堆顶是否到期，再取任务；不能把将来的任务先弹出来丢掉。
• limit=0 返回空列表，负值拒绝。NaN 等无法形成可靠全序的时间值也拒绝。
• 重复待执行标识不更新原任务时间；若需要重调度，应另外定义取消或修改操作，不能悄悄改变题目契约。
• 出堆和移除去重标识在一次临界区完成，业务执行在外部进行。

提交的堆操作为 O(log n)，集合判重平均 O(1)；领取 k 条的堆操作为 O(k log n)。这里只保证内存中的领取不重复，不保证任务执行成功或跨进程重启后恢复。若要求失败重试与可靠执行，需要额外的领取租约、确认状态和持久化。

参考资料