如何在考虑函数执行耗时的前提下,让线程相对精准地按固定间隔(如3秒)执行?
问题原因
你现有的实现是在业务函数执行完成后才启动下一次3秒倒计时,因此两次函数启动的间隔等于「函数执行耗时+3秒」,才会出现4~5.5秒的实际间隔。要实现精准的3秒调用间隔,核心是把计时逻辑和业务函数执行逻辑解耦,不受函数执行耗时影响。
解决方案
方案1:单线程串行执行(适用于要求func1串行运行的场景)
该方案提前计算好每一次的执行时间点,每次执行完函数后仅等待到下一个时间点即可,无需等满3秒,两次func1启动的间隔稳定在3秒左右。
import datetime as dt import time def func1(): # 替换为你的实际业务逻辑,此处模拟1~2.5秒执行耗时 time.sleep(1 + 1.5 * (dt.datetime.now().microsecond / 1000000)) def run_scheduler(interval=3, run_count=5): num = run_count next_run_time = time.time() while num > 0: # 计算需要等待的时间,到点立即执行 wait_seconds = next_run_time - time.time() if wait_seconds > 0: time.sleep(wait_seconds) # 先更新下一次执行时间,避免被函数耗时影响 next_run_time += interval # 执行逻辑 current_time = dt.datetime.now() print(f'count: {num}, time now: {current_time}') func1() num -= 1 if __name__ == "__main__": run_scheduler()
方案2:多线程并行执行(适用于允许多个func1实例同时运行的场景)
该方案把业务函数放到独立线程运行,计时逻辑完全不受函数执行耗时影响,哪怕func1执行耗时超过3秒,也能保证每3秒启动一个新的执行实例。
import datetime as dt import time import threading def func1(): # 替换为你的实际业务逻辑,此处模拟1~2.5秒执行耗时 time.sleep(1 + 1.5 * (dt.datetime.now().microsecond / 1000000)) def run_scheduler(interval=3, run_count=5): num = run_count next_run_time = time.time() while num > 0: wait_seconds = next_run_time - time.time() if wait_seconds > 0: time.sleep(wait_seconds) next_run_time += interval # 业务函数放到独立线程执行,不阻塞计时 threading.Thread(target=execute_task, args=(num,), daemon=True).start() num -= 1 def execute_task(count): current_time = dt.datetime.now() print(f'count: {count}, time now: {current_time}') func1() if __name__ == "__main__": run_scheduler()
误差说明
两种方案的误差均来自操作系统的线程调度精度,普通消费级Windows/macOS/Linux系统下误差通常在10ms以内,完全符合常规业务需求。
内容的提问来源于stack exchange,提问作者QuestionStudent
相关产品推荐
相关产品推荐

