Tenacity重试直到成功
你给出的这段代码：
@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或超过最大超时时间为止，每次重试前记录状态。
如果你想，我可以帮你画一个流程图或写成伪代码来加深理解。