如何高效使用Python schedule模块处理API调用与定时Token更新?
问题解决方案
核心问题分析
你的代码里schedule_run()函数中的while 1是阻塞式循环,一旦执行这个函数,程序会一直卡在循环里处理定时任务,永远不会走到后面的apicall()调用。
高效处理方案:用多线程分离定时任务与API调用
把定时更新Token的任务放到子线程中运行,主线程就能正常处理API调用逻辑,同时后台自动维护Token的更新。
改进后的代码示例
import schedule import time import threading # 全局Token变量,加锁保证线程安全 token = "" token_lock = threading.Lock() def generate_token(): global token # 替换成实际生成Token的逻辑 new_token = f"valid_token_{int(time.time())}" with token_lock: token = new_token print(f"已更新Token: {token}") def apicall(): with token_lock: current_token = token if not current_token: print("错误:Token未初始化") return # 替换成实际API调用逻辑 print(f"使用Token调用API: {current_token}") def schedule_run(): # 启动时先生成一次Token,避免初始为空 generate_token() # 每10分钟执行一次Token更新 schedule.every(10).minutes.do(generate_token) while True: schedule.run_pending() time.sleep(1) if __name__ == "__main__": # 启动定时任务后台线程,设置daemon=True让主线程退出时自动终止该线程 scheduler_thread = threading.Thread(target=schedule_run, daemon=True) scheduler_thread.start() # 主线程处理API调用(这里模拟用户触发场景,实际可替换为Web接口、业务逻辑调用等) while True: apicall() time.sleep(5) # 模拟每5秒发起一次API调用,仅作示例
关键改进点说明
- 多线程分离:通过
threading.Thread把定时任务放到后台线程,主线程不再被阻塞,能正常处理API调用逻辑。 - 线程安全:用
threading.Lock保护全局token变量,避免更新和读取操作同时执行引发的竞态问题。 - 初始Token生成:启动定时任务时先执行一次
generate_token(),避免API调用时Token为空的情况。 - 守护线程:设置
daemon=True,确保主线程退出时后台定时任务线程自动终止,无需手动管理。
额外优化建议
- 如果你的API调用是基于Web服务(比如Flask/Django),只需把
apicall改成接口视图函数,主线程处理HTTP请求即可,后台线程仍负责维护Token。 - 可以给Token添加过期时间校验,避免因定时任务异常导致Token失效后仍被错误使用。
内容的提问来源于stack exchange,提问作者Hannah
相关产品推荐
相关产品推荐

