使用Tenacity在Python中征服重试：一个端到端教程
PYTHON TOOLBOX（Python工具箱）
通过强大的重试机制和错误处理技术增强您的Python项目
本文将讨论Tenacity的基本用法和自定义能力。
这个实用的Python库提供了一个重试机制。我们还将通过一个实际的例子探讨Tenacity的重试和异常处理能力。
介绍
想象一下，您正在管理数百个网络服务，其中一些位于海外（延迟很高），而其他一些相当古老（并且不太稳定）。您会有什么感觉？
我的同事王就遇到了这样的困境。他告诉我他很沮丧：
每天他需要检查这些远程服务的调用状态，经常遇到超时问题或其他异常。故障排除特别具有挑战性。
此外，大部分客户端代码都是由他的前任编写的，这使得进行大规模重构非常困难。因此，这些服务必须继续按原样运行。
如果有一种方法可以在发生异常后自动重新连接这些远程调用就好了。王含着泪看着我。
我向他保证这没有问题，并向他介绍了我的工具箱中的一个新工具：Tenacity。只需一个装饰器，现有的代码就能获得重试能力。让我们看看如何使用它。
安装和基本用法
由于Tenacity的官方网站只提供了简单的API文档，让我们从库的安装和一些基本用法开始。
安装
如果您使用pip，只需运行以下命令：
python -m pip install tenacity如果您使用Anaconda，默认通道中没有Tenacity，因此您需要从conda-forge安装：
conda install -c conda-forge tenacity基本用法
安装Tenacity后，让我们来看看库的一些基本用法。
只需添加一个@retry装饰器，您的代码就具备了重试能力：
@retry()async def coro_func(): pass如果您希望您的代码在尝试一定次数后停止重试，您可以这样编写：
@retry(stop=stop_after_attempt(5))async def coro_func(): pass当然，为了避免频繁重试可能耗尽连接池，我建议在每次重试之前添加等待时间。例如，如果您希望在每次连接之前等待2秒钟：
@retry(wait=wait_fixed(2))async def coro_func(): pass虽然文档中没有提到，但我建议每次重试时等待时间比上一次多一秒，以减少资源浪费：
@retry(wait=wait_incrementing(start=1, increment=1, max=5))async def coro_func(): pass最后，如果重试是由方法中抛出的异常引起的，最好将异常重新抛出。这样在调用该方法时可以进行更灵活的异常处理：
@retry(reraise=True, stop=stop_after_attempt(3))async def coro_func(): pass高级功能：自定义回调
除了一些常见用例外，您可以根据方法执行结果或在执行之前打印方法调用参数等自定义重试判断逻辑。
在这种情况下，我们可以使用自定义回调进行自定义。
有两种方法可以扩展自定义回调函数：
一种是文档推荐的方法：编写扩展方法。
当执行时，此方法将接收一个RetryCallState实例作为参数。
通过这个参数，我们可以获取包装方法、方法调用的参数、返回结果和任何抛出的异常。
例如，我们可以使用这种方法来判断一个方法的返回值，如果返回值是偶数，则重试：
from tenacity import *def check_is_even(retry_state: RetryCallState): if retry_state.outcome.exception(): return True return retry_state.outcome.result() % 2 == 0当然，在做出这个判断之前，如果抛出了异常，则直接重试。
如果需要在扩展方法中传递额外的参数，可以在扩展方法外部添加一个包装器。
例如，这个包装器将传递一个logger参数。当重试次数超过两次时，它将把重试次数、方法名称和方法参数打印到日志中：
def my_before_log(logger: Logger): def my_log(retry_state: RetryCallState): fn = retry_state.fn args = retry_state.args attempt = retry_state.attempt_number if attempt > 2: logger.warning(f"Start retry method {fn.__name__} with args: {args}") return my_log真实世界的网络示例
最后，为了深入了解如何在项目中使用Tenacity，我将以一个远程客户端项目为例，演示如何集成Tenacity的强大功能。
该项目将模拟访问一个HTTP服务，并根据返回的状态码决定是否重试。
当然，为了避免因连接等待时间过长而浪费服务器资源，我还会为每个请求添加一个2秒的超时时间。如果超时发生，将进行重试。
在开始编码之前，我将实现几个扩展方法。其中一个方法是判断一个方法的重试次数是否超过两次，并在日志中打印警告信息：
另一个扩展方法是判断返回的状态码。如果状态码大于300，则重试。当然，超时也会触发重试。
接下来，我们有远程调用方法的实现。在编写方法之后，记得添加Tenacity的重试装饰器。我在这里使用的策略是最多重试20次，在每次重试之前等待上一次重试的时间加1秒。
当然，别忘了添加我们刚刚实现的两个扩展方法：
任务完成！这不是超级简单吗？
结论
我很高兴能帮助Wang解决另一个问题。
通过使用Tenacity，我们可以轻松为现有代码添加各种重试机制，从而增强程序的健壮性和自恢复能力。
如果这个库能帮助您解决问题，我会更加高兴。随时留下评论并讨论。
除了改进代码执行速度和性能之外，使用各种工具提高工作效率也是一种性能提升：
Peng Qian
Python工具箱
查看列表4个故事 
使用我的推荐链接加入小猪AI – Peng Qian
作为小猪AI会员，您的会员费的一部分将用于支付您阅读的作者，并且您可以完全访问每个故事…
小猪AI.com
