你给出的这段代码:
@tenacity.retry(retry=tenacity.retry_if_result(lambdais_acquired:bool(notis_acquired)),stop=tenacity.stop_after_delay(timeout),before=tries)是Tenacity库中的一个重试装饰器,用于自动重试某个函数,直到满足特定条件为止。
🔍 逐部分解释
1.@tenacity.retry(...)
这是 Tenacity 提供的装饰器,用于包装一个函数,使其在失败时自动重试。
2.retry=tenacity.retry_if_result(lambda is_acquired: bool(not is_acquired))
- 作用:只有当函数的返回值为
False时,才触发重试。 retry_if_result(...)是 Tenacity 提供的一个条件重试器。lambda is_acquired: bool(not is_acquired)是一个匿名函数,等价于:
defcheck(is_acquired):returnnotis_acquired# 即如果未获取锁,就重试✅总结:如果函数返回False,就重试;返回True,就停止重试。
3.stop=tenacity.stop_after_delay(timeout)
- 作用:设置最大重试时间为
timeout秒(由函数参数传入,默认是 300 秒)。 - 超过这个时间后,即使条件仍不满足,也停止重试。
4.before=tries
- 作用:每次重试前调用
tries对象(是一个RetryState实例),通常用于记录重试次数或日志。 - 你可以理解为:每次重试前,执行
tries(),比如打印日志或更新状态。
🧪 举个例子
假设你有一个函数try_lock(),它尝试获取一个分布式锁,返回True表示成功,False表示失败。
importtenacityimporttimeclassRetryState:def__init__(self):self.tries=0def__call__(self,retry_state):self.tries+=1print(f"[RetryState] 第{self.tries}次尝试...")# 模拟一个锁获取函数,前 4 次失败,第 5 次成功call_count=0deftry_lock():globalcall_count call_count+=1success=call_count>=5print(f"[try_lock] 第{call_count}次调用,结果:{'成功'ifsuccesselse'失败'}")returnsuccess# 使用 tenacity 重试机制@tenacity.retry(retry=tenacity.retry_if_result(lambdax:notx),# 返回 False 就重试stop=tenacity.stop_after_delay(10),# 最多重试 10 秒before=RetryState()# 每次重试前调用)defacquire_lock():returntry_lock()# 调用result=acquire_lock()print("最终结果:",result)✅ 输出示例:
[RetryState] 第 1 次尝试... [try_lock] 第 1 次调用,结果:失败 [RetryState] 第 2 次尝试... [try_lock] 第 2 次调用,结果:失败 [RetryState] 第 3 次尝试... [try_lock] 第 3 次调用,结果:失败 [RetryState] 第 4 次尝试... [try_lock] 第 4 次调用,结果:失败 [RetryState] 第 5 次尝试... [try_lock] 第 5 次调用,结果:成功 最终结果: True✅ 总结一句话
这个装饰器的作用是:
每隔一段时间重试一次函数,直到它返回
True或超过最大超时时间为止,每次重试前记录状态。
如果你想,我可以帮你画一个流程图或写成伪代码来加深理解。