文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

Python系列:如何提高python程序代码的健壮性

2024-11-30 04:57

关注

可能大多数人会想到使用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()

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

(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()

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

(3) wait_random_min和wait_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()

(4) wait_exponential_multiplier和wait_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()

(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()

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

(6) 其他参数

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

来源:Python都知道内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯