如何提高python程序代码的健壮性

简介: 在编程的时候,我们难免会遇到一些不可靠的情况,比如网络请求失败,数据库连接超时等等。这些不确定性会让我们的程序容易出现各种错误和异常。那么如何来增加程序的容错性和健壮性呢?可能大多数人会想到使用try except来进行异常捕捉进行失败重试(Retry)。虽然try-escept一个非常常见和有效的方式来增强程序稳定性,但是可能一不小心就会造成栈溢出。所以接下来我就来介绍一个另外的一个专门用于失败重试的库:retrying。

在编程的时候,我们难免会遇到一些不可靠的情况,比如网络请求失败,数据库连接超时等等。这些不确定性会让我们的程序容易出现各种错误和异常。那么如何来增加程序的容错性和健壮性呢?

可能大多数人会想到使用try  except来进行异常捕捉进行失败重试(Retry)。虽然try-escept一个非常常见和有效的方式来增强程序稳定性,但是可能一不小心就会造成栈溢出。

所以接下来我就来介绍一个另外的一个专门用于失败重试的库:retrying

定义

在Python生态中,retrying库提供了非常便捷的装饰器和函数来帮助我们轻松添加失败重试机制。它可以自定义重试策略、停止条件、等待间隔等,对各种异常进行捕捉处理。使用retrying可以大大减少我们重复编写失败重试轮询的代码量。

1.下载retrying

pip install retrying

2.无参数重试

我们可以直接在函数上使用装饰器@retry来进行失败重试

import retrying
@retry
def func():
    for item in range(0,100):
        result=item / 0
        print(result)
        return result

func()

但是这种方式并不建议使用,就像上面的代码,我们都知道0作为除数就会报错,在上面的func函数中,因为加了@retry装饰器进行失败重试,这样就就会进入一个死循环一直失败一直重试。

所以我们在进行失败重试的时候最好是需要加上一些参数来限制失败重试。

3.有参数重试

(1)stop_max_attempt_number

在retry中传入stop_max_attempt_number参数后可以指定失败重试的次数

@retry(stop_max_attempt_number=2)
def func():
    print(f"记录失败重试")
    for item in range(0,100):
        result=item / 0
        print(result)
        return result

func()

因为这里我们指定了失败后进行两次重试,如果重试执行两次后还是报错则结束重试,将错误信息抛出来。

1716270626690.jpg

(2)wait_fixed**传入wati_fixed后,可以指定重试的时间

from retrying import retry
import time

# 设置三秒重试一次
@retry(wait_fixed=3000)  
def func():
    print(f"记录失败重试:",time.strftime("%Y-%m-%d %H:%M:%S"))

    result=1 / 0
    print(result)
    return result

func()

1716270669349.jpg

配置重试间隔时间后,成语遇到执行失败或者报错后,就会根据设置的重试时间去进行重试执行

(3)wait_random_minwait_random_max

通常wait_random_min和wait_random_max是一起搭配使用的,可以设置一个重试等待的时间,然后会在设置的时间区间内随机取一个等待时间进行重试

from retrying import retry
import time


@retry(wait_random_min=1000,wait_random_max=9000)
def func():
    print(f"记录失败重试:",time.strftime("%Y-%m-%d %H:%M:%S"))

    result=1 / 0
    print(result)
    return result

func()

1716270702104.jpg

(4)wait_exponential_multiplierwait_exponential_max

官方解释为:以指数的形式产生两次retrying之间的停留时间, 产生的值为2^previous_attempt_number * wait_exponential_multiplier, previous_attempt_number是前面已经retry的次数, 如果产生的这个值超过了wait_exponential_max的大小, 那么之后两个retrying之间的停留值都为wait_exponential_max

通俗来点讲就是每次重试的时间以wait_exponential_multiplier设置的值2,如果重试后还是失败则继续2,直到最后的值等于或则超过wait_exponential_max设置的值后,后面的每一次重试等待时间都是wait_exponential_max设置的值

from retrying import retry
import time

@retry(wait_exponential_multiplier=1000,wait_exponential_max=10000)
def func():
    print(f"记录失败重试:",time.strftime("%Y-%m-%d %H:%M:%S"))

    result=1 / 0
    print(result)
    return result

func()

1716270729554.jpg

(5)wait_func

在前面介绍的参数都是如何配置失败冲重试的等待时间或者重试次数之类的,但是我们不能时时刻刻盯着程序,在程序代码发生错误时我们应该要进行发送短信或者邮件之类的提醒才行

在这里就可以使用到wait_func参数,它接收一个可执行函数,返回一个具体的间隔时间数值,单位ms。接收的函数须接收两个参数:attempt_number当前运行次数,delay_since_first_attempt_ms当前重试机制运行时间(单位ms)

from retrying import retry
import time


def func_demo(attempt_number,delay_since_first_attempt_ms):
    print("函数运行失败后运行该函数")

    if attempt_number == 5:
        print("已经重试失败五次了,开始准备发送提醒")

    if attempt_number == 10:
        print("已经重试失败超10次了,发送邮件给相关人员紧急处理")

    if attempt_number >10:
        print("重试时间过长,做一些其他临时方案进行补救")

    # return一个重试的时间
    return 2000


@retry(wait_func=func_demo)
def func():
    print(f"记录失败重试:",time.strftime("%Y-%m-%d %H:%M:%S"))

    result=1 / 0

    return result

func()

1716270766161.jpg

使用wait_func通过调用其他可执行的函数,我们可以借助它来做一些临时的补救措施,避免程序一直无法运行而产生的影响。

(6)其他参数

在retry中还存在有很多参数,有兴趣的小伙伴可以去详细了解下

1716270780078.jpg

  • stop_max_attempt_number:在停止之前尝试的最大次数,最后一次如果还是有异常则会抛出异常,停止运行,默认为5次
  • stop_max_delay:最大延迟时间,大概意思就是:如果调用的函数出现异常,那么就会重复调用这个函数,最大调用时间,默认为100毫秒
  • wait_fixed:两次调用方法期间停留时长, 如果出现异常则会一直重复调用,默认 1000毫秒
  • wait_random_min:在两次调用方法停留时长,停留最短时间,默认为0
  • wait_random_max:在两次调用方法停留时长,停留最长时间,默认为1000毫秒
  • wait_incrementing_increment:每调用一次则会增加的时长,默认 100毫秒
  • wait_exponential_multiplierwait_exponential_max:以指数的形式产生两次「retrying」之间的停留时间,产生的值为2^previous_attempt_number * wait_exponential_multiplier,previous_attempt_number是前面已经「retry」的次数,如果产生的这个值超过了wait_exponential_max的大小,那么之后两个「retrying」之间的停留值都为wait_exponential_max
  • retry_on_exception: 指定一个函数,如果此函数返回指定异常,则会重试,如果不是指定的异常则会退出
  • retry_on_result:指定一个函数,如果指定的函数返回True,则重试,否则抛出异常退出
  • wrap_exception:参数设置为True/False,如果指定的异常类型,包裹在RetryError中,会看到RetryError和程序抛的Exception error
  • stop_func: 每次抛出异常时都会执行的函数,如果和stop_max_delay、stop_max_attempt_number配合使用,则后两者会失效 (指定的stop_func会有两个参数:attempts, delay)
  • wait_func:和stop_func用法差不多。


作者:大话性能

链接:https://juejin.cn/post/7371000336683532338

目录
打赏
0
2
2
0
440
分享
相关文章
|
2月前
|
时间序列异常检测:MSET-SPRT组合方法的原理和Python代码实现
MSET-SPRT是一种结合多元状态估计技术(MSET)与序贯概率比检验(SPRT)的混合框架,专为高维度、强关联数据流的异常检测设计。MSET通过历史数据建模估计系统预期状态,SPRT基于统计推断判定偏差显著性,二者协同实现精准高效的异常识别。本文以Python为例,展示其在模拟数据中的应用,证明其在工业监控、设备健康管理及网络安全等领域的可靠性与有效性。
585 13
时间序列异常检测:MSET-SPRT组合方法的原理和Python代码实现
【Azure Developer】分享两段Python代码处理表格(CSV格式)数据 : 根据每列的内容生成SQL语句
本文介绍了使用Python Pandas处理数据收集任务中格式不统一的问题。针对两种情况:服务名对应多人拥有状态(1/0表示),以及服务名与人名重复列的情况,分别采用双层for循环和字典数据结构实现数据转换,最终生成Name对应的Services列表(逗号分隔)。此方法高效解决大量数据的人工处理难题,减少错误并提升效率。文中附带代码示例及执行结果截图,便于理解和实践。
Python中main函数:代码结构的基石
在Python中,`main`函数是程序结构化和模块化的重要组成部分。它实现了脚本执行与模块导入的分离,避免全局作用域污染并提升代码复用性。其核心作用包括:标准化程序入口、保障模块复用及支持测试驱动开发(TDD)。根据项目复杂度,`main`函数有基础版、函数封装版、参数解析版和类封装版四种典型写法。 与其他语言相比,Python的`main`机制更灵活,支持同一文件作为脚本运行或模块导入。进阶技巧涵盖多文件项目管理、命令行参数处理、环境变量配置及日志集成等。此外,还需注意常见错误如全局变量污染和循环导入,并通过延迟加载、多进程支持和类型提示优化性能。
44 0
|
2月前
|
[oeasy]python074_ai辅助编程_水果程序_fruits_apple_banana_加法_python之禅
本文回顾了从模块导入变量和函数的方法,并通过一个求和程序实例,讲解了Python中输入处理、类型转换及异常处理的应用。重点分析了“明了胜于晦涩”(Explicit is better than implicit)的Python之禅理念,强调代码应清晰明确。最后总结了加法运算程序的实现过程,并预告后续内容将深入探讨变量类型的隐式与显式问题。附有相关资源链接供进一步学习。
42 4
实战指南:通过1688开放平台API获取商品详情数据(附Python代码及避坑指南)
1688作为国内最大的B2B供应链平台,其API为企业提供合法合规的JSON数据源,直接获取批发价、SKU库存等核心数据。相比爬虫方案,官方API避免了反爬严格、数据缺失和法律风险等问题。企业接入1688商品API需完成资质认证、创建应用、签名机制解析及调用接口四步。应用场景包括智能采购系统、供应商评估模型和跨境选品分析。提供高频问题解决方案及安全合规实践,确保数据安全与合法使用。立即访问1688开放平台,解锁B2B数据宝藏!
【Azure Developer】编写Python SDK代码实现从China Azure中VM Disk中创建磁盘快照Snapshot
本文介绍如何使用Python SDK为中国区微软云(China Azure)中的虚拟机磁盘创建快照。通过Azure Python SDK的Snapshot Class,指定`location`和`creation_data`参数,使用`Copy`选项从现有磁盘创建快照。代码示例展示了如何配置Default Azure Credential,并设置特定于中国区Azure的`base_url`和`credential_scopes`。参考资料包括官方文档和相关API说明。
|
4月前
|
Python高性能编程:五种核心优化技术的原理与Python代码
Python在高性能应用场景中常因执行速度不及C、C++等编译型语言而受质疑,但通过合理利用标准库的优化特性,如`__slots__`机制、列表推导式、`@lru_cache`装饰器和生成器等,可以显著提升代码效率。本文详细介绍了这些实用的性能优化技术,帮助开发者在不牺牲代码质量的前提下提高程序性能。实验数据表明,这些优化方法能在内存使用和计算效率方面带来显著改进,适用于大规模数据处理、递归计算等场景。
109 5
Python高性能编程:五种核心优化技术的原理与Python代码
|
5月前
|
课程设计项目之基于Python实现围棋游戏代码
游戏进去默认为九路玩法,当然也可以选择十三路或是十九路玩法 使用pycharam打开项目,pip安装模块并引用,然后运行即可, 代码每行都有详细的注释,可以做课程设计或者毕业设计项目参考
106 33
在Python程序中实现LevelDB的海量key的分批次扫描
通过本文的步骤,您可以在Python程序中实现对LevelDB海量key的分批次扫描。这样不仅能够有效地管理大规模数据,还可以避免一次性加载过多数据到内存中,提高程序的性能和稳定性。希望这篇指南能为您的开发工作提供实用的帮助。
105 28
【Azure Developer】Python代码调用Graph API将外部用户添加到组,结果无效,也无错误信息
根据Graph API文档,在单个请求中将多个成员添加到组时,Python代码示例中的`members@odata.bind`被错误写为`members@odata_bind`,导致用户未成功添加。
72 10

热门文章

最新文章

AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等