Token需在剩余寿命约20%时主动串行刷新,用threading.Lock保护并以UTC时间戳记录有效期;刷新失败保留旧Token并退避重试;用threading.Timer调度,回调中校验有效性后交线程池执行带超时的HTTP请求。

Token过期前必须主动刷新,不能等请求失败再重试
第三方API的Token通常有固定有效期(比如1小时),靠单次请求返回的expires_in值计算绝对过期时间不可靠——服务器时钟偏差、网络延迟、本地时间不准都会导致误判。更稳妥的做法是:在Token剩余寿命约20%时触发刷新,且刷新动作必须串行化,避免多个线程同时发起刷新请求造成重复或冲突。
实操建议:
- 用
threading.Lock保护刷新逻辑,确保同一时刻最多一个线程执行refresh_token() - 把Token和其
expires_at时间戳封装进一个全局可变对象(如dict或dataclass),所有线程读取都基于它 - 不要依赖
time.time() + expires_in,改用datetime.now(timezone.utc)记录获取时刻,后续统一用UTC比较 - 刷新失败时需保留旧Token继续使用,并设置退避重试(如30秒后再次尝试),而不是直接抛异常中断服务
用threading.Timer实现轻量级定时轮询,别用while True + time.sleep
在主线程里写while True: refresh_if_needed(); time.sleep(60)看似简单,但会阻塞整个程序,且无法响应信号或优雅退出。而threading.Timer是单次触发、自动销毁的,更适合“到期检查→必要时刷新→重新设下一次Timer”的模式。
实操建议:
- 每次成功刷新后,立即创建一个新的
threading.Timer,延迟为max(30, (expires_at - now).total_seconds() * 0.8),确保留出安全余量 - Timer回调函数内先判断当前Token是否仍有效(防止上一轮Timer延迟触发时Token已被其他线程刷新),再决定是否真正执行刷新
- 务必调用
timer.cancel()在程序退出前清理所有待触发Timer,否则可能引发RuntimeError: can't register atexit after shutdown - 避免在Timer回调中做耗时操作(如HTTP请求),应交由线程池处理,Timer只负责调度
HTTP请求必须带超时,且刷新失败不阻塞业务线程
Token刷新本质是一次HTTP请求,若第三方API响应慢或不可达,requests.post(...)默认会无限等待,导致轮询线程卡死,进而影响整个Token状态判断。业务线程(比如处理用户请求的Flask路由)更不能被拖慢。
实操建议:
- 所有刷新请求必须显式设置
timeout=(3, 10)(3秒连通,10秒读取) - 刷新逻辑放在
concurrent.futures.ThreadPoolExecutor中提交,主轮询线程只管触发和监听完成,不等待结果 - 业务线程每次发API请求前,仅做轻量检查:
if token_info['expires_at'] ,不在此刻刷新 - 捕获
requests.exceptions.RequestException和json.JSONDecodeError,记录日志并保持旧Token可用
多线程共享Token状态时,避免用global变量硬编码
直接写global access_token, expires_at不仅难测试,还会在模块重载或单元测试中引发状态污染。更合理的方式是把Token容器设计成可注入、可替换的对象。
实操建议:
- 定义一个
TokenManager类,内部用threading.RLock(支持同一线程多次acquire)保护读写 - 提供
get_token()(无锁读)、set_token(token, expires_in)(带锁写)、needs_refresh()(只读判断)三个接口 - 所有业务代码通过依赖注入方式拿到该实例,而非import全局变量
- 在测试中可轻松Mock
TokenManager,比如强制返回已过期Token来验证刷新逻辑
真正麻烦的不是启动一个定时线程,而是让所有线程对“当前Token是否有效”达成一致认知,且不因刷新动作互相干扰。锁粒度、时间精度、错误隔离,这三处稍有疏忽,就会出现500错误频发却查不出原因的情况。
大量免费API接口:立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游,为数据API应用能力赋能!











