如何在Python的while循环中同时运行连续函数与周期性函数
实现持续函数与周期性函数并行执行
要实现你需要的效果,不能直接按示例代码编写——如果function_runs_continiously()是阻塞型任务,function_runs_every_minute()会一直等待它执行完毕才运行,根本无法实现周期性执行。下面提供两种实用方案:
方案一:非线程时间检查法(适合短任务场景)
如果function_runs_continiously()每次执行耗时很短(非长时间阻塞),可以通过记录时间戳,在每次执行完持续函数后检查是否到了周期任务的执行时间:
import time def function_runs_continiously(): # 模拟持续执行的短任务,执行后立即返回 print("持续执行中...") time.sleep(0.1) # 模拟工作耗时,避免CPU占满 def function_runs_every_minute(): print("执行周期性任务...") # 初始化上次执行周期任务的时间 last_run_time = time.time() interval = 60 # 周期为60秒 while True: function_runs_continiously() # 检查是否达到执行周期 current_time = time.time() if current_time - last_run_time >= interval: function_runs_every_minute() last_run_time = current_time
方案二:多线程法(适合阻塞型任务场景)
如果function_runs_continiously()是长时间阻塞的任务(比如持续监听消息、处理数据流),需要用多线程让两个任务真正并行:
import time import threading def function_runs_continiously(): while True: print("持续执行中...") time.sleep(0.1) # 模拟阻塞型持续任务 def function_runs_every_minute(): while True: print("执行周期性任务...") time.sleep(60) # 间隔60秒执行一次 # 创建周期任务线程,设置为守护线程(主线程结束时自动退出) periodic_thread = threading.Thread(target=function_runs_every_minute, daemon=True) periodic_thread.start() # 主线程运行持续任务 function_runs_continiously()
注意事项
- 多线程场景下,如果两个函数需要操作共享资源(比如全局变量、文件),必须用
threading.Lock()加锁,避免数据竞争导致的错误。 - 非线程法的精度依赖于
function_runs_continiously()的执行频率,如果它每次执行间隔太长,周期任务的执行时间会有偏差。
内容的提问来源于stack exchange,提问作者lucmad
相关产品推荐
相关产品推荐

