Python如何仅用单个额外线程调度定时短任务 避免创建数百个线程

实现方案 #

完全基于Python标准库实现，仅使用1个额外线程完成所有定时任务调度，支持任务取消，满足你的需求：

核心思路 #

用最小堆（heapq）存储所有待执行任务（按触发时间排序），搭配threading.Condition实现等待唤醒逻辑，所有任务调度逻辑仅运行在单个后台线程中，不需要为每个任务新建线程。

完整实现代码 #

import threading
import time
import heapq
from dataclasses import dataclass
from typing import Callable, Any

@dataclass(order=True)
class ScheduledTask:
    trigger_time: float
    canceled: bool = False
    args: Any = None
    func: Callable = None

class SingleThreadScheduler:
    def __init__(self):
        self._task_heap = []
        self._cond = threading.Condition(threading.Lock())
        self._running = True
        # 启动唯一的调度线程
        self._thread = threading.Thread(target=self._scheduler_loop, daemon=True)
        self._thread.start()
    
    def _scheduler_loop(self):
        while self._running:
            with self._cond:
                # 等待直到有任务
                while not self._task_heap:
                    self._cond.wait()
                
                # 取最近要触发的任务
                now = time.time()
                task = self._task_heap[0]
                if task.canceled:
                    heapq.heappop(self._task_heap)
                    continue
                
                delay = task.trigger_time - now
                if delay <= 0:
                    # 到执行时间，弹出任务
                    heapq.heappop(self._task_heap)
                    # 释放锁之后再执行任务，避免阻塞新任务提交
                    self._cond.release()
                    try:
                        task.func(*task.args)
                    finally:
                        self._cond.acquire()
                else:
                    # 等待到触发时间，或者有新任务/取消操作唤醒
                    self._cond.wait(delay)
    
    def submit(self, delay: float, func: Callable, *args) -> ScheduledTask:
        """提交延迟执行的任务，返回任务对象可用于取消"""
        trigger_time = time.time() + delay
        task = ScheduledTask(trigger_time=trigger_time, func=func, args=args)
        with self._cond:
            heapq.heappush(self._task_heap, task)
            self._cond.notify()
        return task
    
    def cancel_all(self):
        """取消所有待执行任务"""
        with self._cond:
            for task in self._task_heap:
                task.canceled = True
            self._task_heap.clear()
            self._cond.notify()
    
    def shutdown(self):
        """关闭调度器"""
        with self._cond:
            self._running = False
            self._cond.notify()
        self._thread.join()

业务逻辑对接示例 #

和你原来的代码逻辑完全对齐，替换后不需要修改业务流程：

# 初始化调度器，全局仅一个实例，仅启动1个后台线程
scheduler = SingleThreadScheduler()
tasks = []  # 存储任务用于取消，对应你原来的timers变量

while True:
    # 先取消上一轮未执行的任务（如果有需要）
    # scheduler.cancel_all()
    tasks.clear()
    for i in range(20):
        # 提交延迟任务，不会新建线程
        task = scheduler.submit(i * 0.010, do_it, i)
        tasks.append(task)
    wait_for_something_else() # 这部分逻辑和原来完全一致

方案说明 #

• 线程开销极低：整个调度逻辑仅占用1个常驻后台线程，即使同时有上千个待执行任务也不会新增线程
• 时间精度足够：你需求的10ms级延迟调度完全可以满足，do_it执行耗时远低于0.1ms，不会对后续任务调度产生影响
• 完全标准库实现：不需要安装任何第三方依赖
• 支持任务取消：可以单独取消某个任务，也可以批量取消所有待执行任务，和你原来的需求一致
• 无空转消耗：Condition的等待逻辑会主动让出CPU，不会造成不必要的资源占用

内容的提问来源于stack exchange，提问作者Basj