Python func_timeout模块与retrying模块

安装：

conda install func_timeout retrying

简单版本

用func_timeout做超时（他就是个wrapper，也可以自己写一个，但是就重复造轮子了）

import func_timeout

@func_set_timeout(20)
def my_func():
    pass

retrying根据的超时异常func_timeout.exceptions.FunctionTimedOut进行重试（也可以自己写自己的Exception，包装异常信息，方便后面排查）

import retrying
import func_timeout

def is_need_retry(exception: Exception) -> bool:
    return isinstance(exception, func_timeout.exceptions.FunctionTimedOut)

#此处的wait_fixed指的是被retrying修饰的函数每次重试的间隔时间，区别于上面的timeout
@retrying.retry(retry_on_exception=is_need_retry, stop_max_attempt_number=20, wait_fixed=2000)
def main():
    my_func()

测试：

import random
import time

import func_timeout
import retrying
from func_timeout import func_set_timeout


class my_exception(Exception):
    def __init__(self, message):
        self.message = message

    def __str__(self):
        message = 'Sleep timeout: "{}"'.format(self.message)
        print(message)
        return message


def sleep(sleep_time: int):
    timeout = 3

    @func_set_timeout(timeout)
    def do_sleep():
        print('sleep time ', sleep_time)
        time.sleep(sleep_time)

    try:
        do_sleep()
        print('sleep finish')
    except func_timeout.exceptions.FunctionTimedOut:
        print('sleep timeout')
        raise my_exception(sleep_time)


def sleep_timeout(exception: Exception) -> bool:
    return isinstance(exception, my_exception)


@retrying.retry(retry_on_exception=sleep_timeout, stop_max_attempt_number=10, wait_fixed=2000)
def test():
    print('start @ ', time.asctime())
    sleep(random.randint(2, 5))
    print('end')


if __name__ == '__main__':
    random.seed(1234)
    test()

start @ Wed Jan 20 22:15:01 2021
sleep time 5
sleep timeout
start @ Wed Jan 20 22:15:06 2021
sleep time 2
sleep finish
end

参考：

https://www.cnblogs.com/mangM/p/11207202.html
https://www.jb51.net/article/164084.htm