python 超时重试方法

在应用中,有时候会 依赖第三方模块执行方法,比如调用某模块的上传下载,数据库查询等操作的时候,如果出现网络问题或其他问题,可能有超时重新请求的情况;

目前的解决方案有

1. 信号量,但不支持window;

2.多线程,但是 如果是大量的数据重复操作尝试,会出现线程管理混乱,开启上万个线程的问题;

3.结合采用 eventlet 和 retrying模块 (eventlet 原理尚需深入研究)

下面的方法实现:超过指定时间重新尝试某个方法

# -*- coding: utf-8 -*-
import random
import time

import eventlet
from retrying import retry

eventlet.monkey_patch()

class RetryTimeOutException(Exception):
    def __init__(self, *args, **kwargs):
        pass

def retry_if_timeout(exception):
    """Return True if we should retry (in this case when it‘s an IOError), False otherwise"""
    return isinstance(exception, RetryTimeOutException)

def retry_fun(retries=3, timeout_second=2):
    """
    will retry ${retries} times when process time beyond ${timeout_second} ;
    :param retries: The retry times
    :param timeout_second: The max process time
    """

    def retry_decor(func):
        @retry(stop_max_attempt_number=retries, retry_on_exception=retry_if_timeout)
        def decor(*args, **kwargs):
            print("In retry method..")
            pass_flag = False
            with eventlet.Timeout(timeout_second, False):
                r = func(*args, **kwargs)
                pass_flag = True
                print("Success after method.")
            if not pass_flag:
                raise RetryTimeOutException("Time out..")
            print("Exit from retry.")
            return r

        return decor

    return retry_decor

def do_request():
    print("begin request...")
    sleep_time = random.randint(1, 4)
    print("request sleep time: %s." % sleep_time)
    time.sleep(sleep_time)
    print("end request...")
    return True

@retry_fun(retries=3)
def retry_request():
    r = do_request()
    print(r)

if __name__ == ‘__main__‘:
    retry_request()

  参考:

安装依赖模块:pip install retrying eventlet -i https://pypi.tuna.tsinghua.edu.cn/simple/

装饰器用法:https://blog.csdn.net/u013205877/article/details/78872278

retry: https://blog.csdn.net/lxy210781/article/details/95253026

超时:https://blog.csdn.net/yuanpython/article/details/90522567

其他方法:https://www.cnblogs.com/lyxdw/p/10033118.html

原文地址:https://www.cnblogs.com/dasheng-maritime/p/11602921.html

时间: 2024-08-30 15:50:09

python 超时重试方法的相关文章

Volley超时重试机制详解

Volley超时重试机制 基础用法 Volley为开发者提供了可配置的超时重试机制,我们在使用时只需要为我们的Request设置自定义的RetryPolicy即可. 参考设置代码如下: int DEFAULT_TIMEOUT_MS = 10000; int DEFAULT_MAX_RETRIES = 3; StringRequest stringRequest = new StringRequest(Request.Method.GET, url, new Response.Listener<S

dubbo超时重试和异常处理

参考: https://www.cnblogs.com/ASPNET2008/p/7292472.html https://www.tuicool.com/articles/YfA3Ub https://www.cnblogs.com/binyue/p/5380322.html https://blog.csdn.net/mj158518/article/details/51228649 本篇主要记录dubbo中关于超时的常见问题,实现原理,解决的问题以及如何在服务降级中体现作用等. 超时问题

使用 Polly 实现复杂策略(超时重试)

一.背景 第一次接触 Polly 还是在做某个微服务系统的时候,那时只会使用单一的超时策略与重试策略,更加高级的特性就没有再进行学习了.最近开为某个客户开发 PC 端的上位机的时候,客户有个需求,在发起请求之后如果 5 秒钟没有响应则进行重试,总共可以重试 3 次,如果 3 次请求都未返回数据,就视为请求失败. 关于 Polly 的高级用法可以参考官方的 Wiki 文档即可,国内也有很多优秀的介绍文章,例如 这篇 和 这篇. 二.思路 查阅了 Polly 的官方文档之后,发现 Polly 提供了

Python内建方法

参考: https://docs.python.org/3.4/library/functions.html https://docs.python.org/2/library/functions.html http://blog.csdn.net/jgood/article/details/4371991 以上链接分别为Python官网的3.4版本的内建方法说明.2.X(指2.6和2.7)版本的内建方法说明.以及JGood对2.X版本的内建方法说明的翻译. abs(x) 返回一个数的绝对值.参

Python数据类型及其方法详解

Python数据类型及其方法详解 我们在学习编程语言的时候,都会遇到数据类型,这种看着很基础也不显眼的东西,却是很重要,本文介绍了python的数据类型,并就每种数据类型的方法作出了详细的描述,可供知识回顾. 一.整型和长整型 整型:数据是不包含小数部分的数值型数据,比如我们所说的1.2.3.4.122,其type为"int" 长整型:也是一种数字型数据,但是一般数字很大,其type为"long" 在python2中区分整型和长整型,在32位的机器上,取值范围是-2

Python内置方法的时间复杂度(转)

原文:http://www.orangecube.net/python-time-complexity 本文翻译自Python Wiki本文基于GPL v2协议,转载请保留此协议. 本页面涵盖了Python中若干方法的时间复杂度(或者叫“大欧”,“Big O”).该时间复杂度的计算基于当前(译注:至少是2011年之前)的CPython实现.其他Python的实现(包括老版本或者尚在开发的CPython实现)可能会在性能表现上有些许小小的差异,但一般不超过一个O(log n)项. 本文中,’n’代

Python 抽象工厂方法

有没有好的python UML建模工具?求推荐,除eclipse的插件(因为不喜欢用eclipse).pyNsource用的不是很好,pyUt不全.有没StarUML上的python插件? import abc class AbstractEnemyFactory( object ): __metaclass__ = abc.ABCMeta @abc.abstractmethod def createNinja( self ): pass @abc.abstractmethod def crea

Python中strip方法的妙用

[开胃小菜] 当提到python中strip方法,想必凡接触过python的同行都知道它主要用来切除空格.有以下两种方法来实现. 方法一:用内置函数 #<python> if __name__ == '__main__': str = ' Hello world ' print '[%s]' %str.strip() #</python> 方法二:调用string模块中方法 #<python> import string if __name__ == '__main__

Python设计模式——工厂方法模式(FactoryMethod)

需求:有一个学雷锋活动,有买米和扫地两个内容,参与的人有大学生和社区志愿者,他们各自的方法不一样. 如果用简单工厂模式实现: #encoding=utf-8 __author__ = '[email protected]' class LeiFeng(): def buy_rice(self): pass def sweep(self): pass class Student(LeiFeng): def buy_rice(self): print '大学生帮你买米' def sweep(self