python 超時重試的方法 signal手段

# -*- coding: utf-8 -*-

import random
import signal
import time

from retrying import retry


def retry_when_timeout(retries=3, timeout=2):
    def decor(func):
        def timeout_handle(signum, frame):
            raise RetryTimeOutException("Time out...")

        @retry(stop_max_attempt_number=retries, retry_on_exception=retry_if_timeout)
        def run(*args, **kwargs):
            signal.signal(signal.SIGALRM, timeout_handle)  # 設置信號和回調函數
            signal.alarm(timeout)  # 設置 num 秒的鬧鐘
            print('start alarm signal.')
            r = func(*args, **kwargs)
            print('close alarm signal.')
            signal.alarm(0)  # 關閉鬧鐘
            return r

        return run

    return decor


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)


@retry_when_timeout(retries=3)
def do_process():
    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


if __name__ == "__main__":
    do_process()

  不支持 window.python

相關文章
相關標籤/搜索