Python Tenacity 库教程介绍在编写可靠的分布式系统或网络应用程序时,我们经常需要处理各种故障和错误。Tenacity 是一个 Python 库,旨在简化处理这些故障和错误的过程。 安装要开始使用 Tenacity,首先需要安装它。 可以使用 pip 来安装:bashCopy codepip install tenacity安装完成后,我们就可以在项目中导入 Tenacity 模块并开始使用了。 pythonCopy codeimport tenacity基础用法装饰器Tenacity 最常见的用法是通过装饰器来标记一个函数,使其具有重试的能力。 例如,我们可以指定重试次数和重试间隔:pythonCopy code@tenacity.retry( stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_fixed
不可避免的会用到一些重试功能,比如数据库和网络重连,或者其他的一些异常方法重试等等,有些组件可能自带了重试功能,但有些组件可能没有带就需要我们自己开发了,不过这种组件一般都有开源成熟的方案,所以我们就没必要重新造轮子了,而tenacity github地址: https://github.com/jd/tenacity 安装方式 pip安装: pip install tenacity anaconda安装: conda install - c conda-forge tenacity 简单例子 一个简单的重试功能,如果发生异常,则会一直重试,直到成功: (1)无限重试 @retry def never_give_up_never_surrender return False 如果结果是False就执行重试,重试的间隔是2秒,重试的次数是4 更多例子可参考: https://tenacity.readthedocs.io/en/latest/
Python 有一个第三方库,叫做Tenacity,它实现了一种优雅的重试功能。 以上面爬虫最初的无限重试版本为例,如果想实现遇到异常就重试。 只需要添加两行代码,爬虫的主体函数完全不需要做修改: from tenacity import retry @retry def extract(url): info_json = requests.get json.loads(info_json) data = info_dict['data'] save(data) 甚至重试的时间间隔想指数级递增,代码行数也不需要增加: from tenacity info_dict字典里面没有data这个key 如果只需要在JSON解析错误时重试,由于异常类型为json.decoder.JSONDecodeError,所以就可以通过参数来进行限制: from tenacity Tenacity是我见过的,最 Pythonic ,最优雅的第三方库。 欢迎关注我的公众号:未闻Code(ID:itskingname)
Python 有一个第三方库,叫做Tenacity,它实现了一种优雅的重试功能。 以上面爬虫最初的无限重试版本为例,如果想实现遇到异常就重试。 只需要添加两行代码,爬虫的主体函数完全不需要做修改: from tenacity import retry @retry def extract(url): info_json = requests.get json.loads(info_json) data = info_dict['data'] save(data) 甚至重试的时间间隔想指数级递增,代码行数也不需要增加: from tenacity info_dict字典里面没有data这个key 如果只需要在JSON解析错误时重试,由于异常类型为json.decoder.JSONDecodeError,所以就可以通过参数来进行限制: from tenacity Tenacity是我见过的,最 Pythonic ,最优雅的第三方库。
Tenacity是一个通用的retry库,简化为任何任务加入重试的功能,它实现了几乎我们可以使用到的所有重试场景。 先pip为敬: pip install tenacity 不懂这个库怎么用, 很简单,哦对了,可能还需要你知道装饰器就是那个@啦就够了,上菜! 无条件重试,重试之间无间隔 import tenacity import requests # 直接加上retry装饰器,代码抛出异常会一直重试,直到代码运行成功 @tenacity.retry() 无条件重试,但是在重试之前要等待 3秒: from tenacity import retry , stop_after_attempt , stop_after_delay , wait_fixed 重试5秒后不再重试 from tenacity import retry , stop_after_attempt , stop_after_delay import requests # 指定5s
tenacity tenacity是一个 Apache 2.0授权的通用重试库,自动化测试或者爬虫中,当网络不稳定导致请求超时或者等待条件满足时操作,我们可以通过tenacity实现代码的重试功能。 https://github.com/jd/tenacity pip install tenacity 用法非常简单,直接加上装饰器使用。 重试3次 import tenacity from tenacity import stop_after_attempt @tenacity.retry(stop=stop_after_attempt raise Exception retry_test() 重试10秒 import tenacity from tenacity import stop_after_delay @tenacity.retry raise Exception retry_test() 每隔2秒重试 import tenacity from tenacity import wait_fixed @tenacity.retry
而今天我要给大家介绍的tenacity库,可能是目前Python生态中最好用的错误重试库,下面就让我们一睹其主要功能吧~ 2 tenacity中的常用功能 作为一个第三方Python库,我们可以使用pip install tenacity对其进行安装,安装完成后,下面我们来学习一下tenacity的主要使用方法和特性: 2.1 tenacity的基础使用 tenacity的错误重试核心功能由其retry 装饰器来实现,默认不给retry装饰器传参数时,它会在其所装饰的函数运行过程抛出错误时不停地重试下去,譬如下面这个简单的例子: import random from tenacity import retry ()接受一个整数输入作为「最大重试」的次数: from tenacity import retry, stop_after_attempt @retry(stop=stop_after_attempt tenacity中同样内置了相关的实用功能: 2.6.1 捕捉或忽略特定的错误类型 使用tenacity中的retry_if_exception_type()和retry_if_not_exception_type
这里要给大家介绍的是一个第三方库 - Tenacity (标题中的重试机制并并不准确,它不是 Python 的内置模块,因此并不能称之为机制),它实现了几乎我们可以使用到的所有重试场景,比如: 在什么情况下才进行重试 在使用它之前 ,先要安装它 $ pip install tenacity 1. 最基本的重试 无条件重试,重试之间无间隔 from tenacity import retry @retry def test_retry(): print("等待重试,重试无间隔执行..." raise Exception test_retry() 无条件重试,但是在重试之前要等待 2 秒 from tenacity import retry, wait_fixed @retry(wait 重试后错误重新抛出 当出现异常后,tenacity 会进行重试,若重试后还是失败,默认情况下,往上抛出的异常会变成 RetryError,而不是最根本的原因。
而今天我要给大家介绍的tenacity库,可能是目前Python生态中最好用的错误重试库,下面就让我们一睹其主要功能吧~ 2 tenacity中的常用功能 作为一个第三方Python库,我们可以使用 pip install tenacity对其进行安装,安装完成后,下面我们来学习一下tenacity的主要使用方法和特性: 2.1 tenacity的基础使用 tenacity的错误重试核心功能由其retry 装饰器来实现,默认不给retry装饰器传参数时,它会在其所装饰的函数运行过程抛出错误时不停地重试下去,譬如下面这个简单的例子: import random from tenacity import retry ()接受一个整数输入作为最大重试的次数: from tenacity import retry, stop_after_attempt @retry(stop=stop_after_attempt(3 tenacity中同样内置了相关的实用功能: 2.6.1 捕捉或忽略特定的错误类型 使用tenacity中的retry_if_exception_type()和retry_if_not_exception_type
以 Nginx 的日志为例,这是一条访问日志: 162.158.167.131 - - [11/Aug/2020:06:47:30 +0800] "GET /tags/Tenacity HTTP/1.1 petalbot)" 这条日志包含了很多信息,包括:访问者的 IP 地址:162.158.167.131,访问发起的时间:11/Aug/2020:06:47:30 +0800,具体访问的路径:/tags/Tenacity 我们用一段简单的代码来进行测试: >>> import parse >>> log = '162.158.167.131 - - [11/Aug/2020:06:47:30 +0800] "GET /tags/Tenacity datetime.datetime(2020, 8, 11, 6, 47, 30, tzinfo=<FixedTzOffset +0800 8:00:00>), 'method': 'GET', 'path': '/tags/Tenacity
我们可以通过两种方式实现重试:手动循环重试、使用 tenacity 第三方库(更优雅、更灵活)。 请求最终失败") return None return Noneif __name__ == '__main__': request_with_retry()方案 2:tenacity 库实现企业级重试(推荐)tenacity 是 Python 官方推荐的重试库,支持指数退避重试、异常过滤、最大重试时间、重试日志等高级功能,完美适配企业级场景。 python运行import requestsimport loggingfrom tenacity import retry, stop_after_attempt, wait_exponential ,而 try 超时控制 + 智能重试机制 是 Python 解决该问题的最优方案:try-except 是超时异常捕获的核心,能主动终止阻塞请求,保护程序稳定性;轻量场景用手动循环重试,企业级场景用 tenacity
三、tenacity(重试+结合多线程/异步,保证有序结果)核心特点并非专门的 HTTP 库,而是 重试库,可与 requests、aiohttp、ThreadPoolExecutor 安装pip install tenacity完整实现(结合 ThreadPoolExecutor + 重试+有序结果)import requestsfrom concurrent.futures import [executor.submit(request_api, i) for i in range(TOTAL_REQUESTS)] print("ThreadPoolExecutor + tenacity requests、语法极简、并发效率高快速替换 requests 实现高并发+有序结果httpx异步(asyncio)支持同步/异步、HTTP/2、连接池优化替代 requests 且追求高性能+有序结果tenacity 提交顺序=任务列表顺序=结果列表顺序”,无需手动排序;结合场景选择库: 简单场景:用 grequests(最快上手);高性能场景:用 httpx(支持 HTTP/2);接口不稳定:用 tenacity
https://github.com/kragniz/python-etcd3/blob/master/requirements.txt 关联插件:grpcio>=1.2.0 tenacity coverage>=4.0 cython>=0.27 enum34>=1.0.4 protobuf>=3.5.0.post1 six>=1.10 wheel>=0.29 2.2、搜索etcd3关联插件tenacity A) 下载相关包 https://pypi.org/project/tenacity/ B) 进入homepage地址:https://github.com/jd/tenacity/ C)关联插件包说明:https://github.com/jd/tenacity/blob/master/requirements.txt 关联插件:six setuptools-39.1.0 dnspython-1.15.0 ullib3-1.22 pyDatalog-0.17.1 pbr-4.0.3 tenacity
示例 1:使用 Tenacity 库Tenacity 是一个遵循 Apache 2.0 许可的通用重试库,用 Python 编写,可简化将重试行为添加到几乎任何东西的任务。 要将指数退避添加到您的请求中,您可以使用 tenacity.retry 装饰器。下面的示例使用 tenacity.wait_random_exponential 函数为请求添加随机指数退避。 from openai import OpenAIclient = OpenAI()from tenacity import ( retry, stop_after_attempt, client.completions.create(**kwargs) completion_with_backoff(model="gpt-3.5-turbo-instruct", prompt="Once upon a time,")请注意,Tenacity client.completions.create(**kwargs) completions_with_backoff(model="gpt-3.5-turbo-instruct", prompt="Once upon a time,")与 Tenacity
四、方案4:tenacity(重试+多线程/异步,有序+高可用)特点并非HTTP库,而是重试库,可与任意并发方案(ThreadPoolExecutor、httpx、aiohttp)结合;支持失败自动重试 安装依赖pip install tenacity requests2. 完整实现(多线程+重试+有序结果)import requestsfrom concurrent.futures import ThreadPoolExecutorfrom tenacity import 按Future列表顺序获取结果(保证有序) print("ThreadPoolExecutor + tenacity 有序结果输出:") print("-" * 60) 兼容requestsgrequests语法极简,无需学习新API高性能、支持HTTP/2httpx同步/异步双模式,稳定性强追求代码简洁、结构化并发trioAPI优雅,错误处理友好接口不稳定、需要重试tenacity
常见的老方法 使用第三方库 这类同学会使用一些专业做重试的第三方库,例如tenacity。 详见我的这篇文章:Tenacity——Exception Retry 从此无比简单 手动写装饰器 这类同学会使用装饰器,所以一般会手写装饰器从而复用,例如: def retry(func): def
run: tree -L 3 -a # - name: clean theme cache # run: git rm -f --cached themes/tenacity # run: git submodule deinit themes/tenacity && git rm themes/tenacity # 安装主题 - name : Install Theme run: git submodule add https://github.com/all-smile/tenacity.git themes/tenacity
安装: pip install tenacity 使用: import random from tenacity import retry @retry def do_something_unreliable
错误处理与重试机制 import tenacity @tenacity.retry( stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential (multiplier=1, min=4, max=10), retry=tenacity.retry_if_exception_type((redis.ConnectionError, redis.TimeoutError
allkeys-random:从数据集中【任意选择】 Key 进行删除 no-enviction:禁止驱逐数据 如上图,6.2 后的版本支持通过逐出因子 maxmemory-eviction-tenacity 是否因内存驱逐阻塞严重,可开启 lazyfree-lazy-eviction 来缓解阻塞,如果还是很慢,可升级内存规格 可升级到 7.x 版本的 Redis ,通过 maxmemory-eviction-tenacity