如何优化Python周期性任务代码，实现低至10ms的执行间隔？

我之前碰到过类似的高精度定时需求，要突破100ms的瓶颈，得从系统时钟精度和任务调度逻辑两个核心点入手。咱们一步步拆解解决方案：

1. 为什么原代码只能到100ms？ #

原代码用threading.Event.wait()来控制间隔，但这个方法有两个硬伤：

• 系统时钟限制：比如Windows默认的系统时钟中断间隔是15.6ms，Linux也有基础调度粒度，直接限制了wait的最小精度；
• 未抵消任务耗时：每次任务执行的时间会直接占用间隔时间——比如设10ms间隔，任务执行用了2ms，实际间隔就变成12ms，加上系统调度延迟，很容易飘到100ms级。

2. 关键优化步骤 #

（1）改用绝对时间调度 #

放弃固定间隔等待，转而计算下一次任务执行的绝对时间点，每次循环只等待「当前时间到下一次执行时间」的差值。这样即使任务有耗时，也能保证两次任务启动的间隔稳定。

（2）提升系统时钟精度 #

在Windows上，我们可以调用系统API把时钟精度拉到1ms；Linux下可以尝试设置实时调度策略（需要root权限），让线程获得更高的调度优先级。

（3）确保任务本身足够轻量 #

如果你的任务执行时间本身就超过10ms，再怎么优化也没用——所以要保证execute函数是纯计算或极轻量操作，避免IO、阻塞调用。

3. 优化后的完整代码 #

import threading
import time
from datetime import timedelta
from typing import Callable
from unittest.mock import Mock
import ctypes

# 提升Windows系统时钟精度到1ms（非Windows系统自动忽略）
try:
    winmm = ctypes.WinDLL('winmm.dll')
    winmm.timeBeginPeriod(1)
except (OSError, AttributeError):
    pass

WAIT_TIME_SECONDS = 0.01  # 目标10ms间隔

class PeriodicTask(threading.Thread):
    """高精度周期性任务执行类"""
    def __init__(self, interval: timedelta, execute: Callable[..., None], *args, **kwargs):
        super().__init__()
        assert isinstance(interval, timedelta), "间隔必须是datetime.timedelta类型"
        assert execute is not None, "必须指定要周期性执行的函数"
        self.daemon = False
        self.stopped = threading.Event()
        self.interval_sec = interval.total_seconds()
        self.execute = execute
        self.args = args
        self.kwargs = kwargs

    def stop(self):
        """优雅停止任务"""
        self.stopped.set()
        self.join()

    def run(self):
        """高精度调度逻辑"""
        next_run_time = time.perf_counter() + self.interval_sec
        
        while not self.stopped.wait(max(0.0, next_run_time - time.perf_counter())):
            # 执行目标任务
            self.execute(*self.args, **self.kwargs)
            
            # 更新下一次执行时间：如果任务耗时超过间隔，直接跳到下一个周期
            next_run_time += self.interval_sec
            if next_run_time < time.perf_counter():
                next_run_time = time.perf_counter() + self.interval_sec

if __name__ == "__main__":
    foo = Mock()
    job = PeriodicTask(interval=timedelta(seconds=WAIT_TIME_SECONDS), execute=foo)
    job.start()
    time.sleep(1)
    job.stop()
    
    # 验证执行次数：理论上1秒内应该执行约100次
    print(f"任务执行次数：{foo.call_count}")

4. 额外优化（针对Linux） #

如果在Linux环境下运行，还可以尝试给线程设置实时调度策略（需要root权限），进一步降低调度延迟。在run方法开头添加：

import os
try:
    os.sched_setscheduler(0, os.SCHED_FIFO, os.sched_param(1))
except PermissionError:
    print("提示：需要root权限才能设置实时调度策略")
except AttributeError:
    pass

内容的提问来源于stack exchange，提问作者Wör Du Schnaffzig