python 超时重试方法
生活随笔
收集整理的這篇文章主要介紹了
python 超时重试方法
小編覺得挺不錯的,現在分享給大家,幫大家做個參考.
在應用中,有時候會 依賴第三方模塊執行方法,比如調用某模塊的上傳下載,數據庫查詢等操作的時候,如果出現網絡問題或其他問題,可能有超時重新請求的情況;
目前的解決方案有
信號量,但不支持window;
多線程,但是 如果是大量的數據重復操作嘗試,會出現線程管理混亂,開啟上萬個線程的問題;
結合采用 eventlet 和 retrying模塊 (eventlet 原理尚需深入研究)
下面的方法實現:超過指定時間重新嘗試某個方法
''' 遇到問題沒人解答?小編創建了一個Python學習交流QQ群:579817333 尋找有志同道合的小伙伴,互幫互助,群里還有不錯的視頻學習教程和PDF電子書! ''' # -*- coding: utf-8 -*- import random import timeimport eventlet from retrying import retryeventlet.monkey_patch()class RetryTimeOutException(Exception):def __init__(self, *args, **kwargs):passdef 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 = Falsewith eventlet.Timeout(timeout_second, False):r = func(*args, **kwargs)pass_flag = Trueprint("Success after method.")if not pass_flag:raise RetryTimeOutException("Time out..")print("Exit from retry.")return rreturn decorreturn retry_decordef 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() 與50位技術專家面對面20年技術見證,附贈技術全景圖總結
以上是生活随笔為你收集整理的python 超时重试方法的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: Python文件读写、StringIO和
- 下一篇: python中operator.item