python 自定义超时模块

1. eventlet (简单好用，但要修改函数)

import requests
import time
import eventlet

eventlet.monkey_patch()  # 必须加这条代码

try:
    # Dontla 20200421 超时将抛出异常
    with eventlet.Timeout(2, True):  # 设置超时时间为2秒
        time.sleep(4)
        r = requests.get("http://10.101.35.249/", verify=False)
        print('error')
    print('程序运行未超时！')
except eventlet.timeout.Timeout:
# except:   # （或，两种都行，注意不能用except Exception，因为Exception是异常基类，我们这个超时异常未包含在它里面）
    print('程序运行超时！')

2. 装饰器（自己写），不用修改原来的函数

from threading import Thread
import time
import inspect
import ctypes
import sys

sys.setrecursionlimit(1000000) #不加不知道为啥出现递归错误

class TimeoutException(Exception):
    pass


def _async_raise(tid, exctype):
    """raises the exception, performs cleanup if needed"""
    tid = ctypes.c_long(tid)
    if not inspect.isclass(exctype):
        exctype = type(exctype)
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype))
    if res == 0:
        raise ValueError("invalid thread id")
    elif res != 1:
        # """if it returns a number greater than one, you're in trouble,
        # and you should call it again with exc=NULL to revert the effect"""
        ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)
        raise SystemError("PyThreadState_SetAsyncExc failed")


def stop_thread(thread):
    _async_raise(thread.ident, SystemExit)


def timelimited(timeout):
    def decorator(function):
        def decorator2(*args, **kwargs):

            class TimeLimited(Thread):
                def __init__(self, _error=None, ):
                    Thread.__init__(self)
                    self._error = _error

                def run(self):
                    try:
                        self.result = function(*args, **kwargs)
                    except Exception as e:
                        self._error = str(e)

                def _stop(self):
                    if self.is_alive():
                        stop_thread(self)

            t = TimeLimited()
            t.start()
            t.join(timeout)

            if isinstance(t._error, TimeoutException):
                t._stop()
                raise TimeoutException('timeout for %s' % (repr(function)))

            if t.is_alive():
                t._stop()
                raise TimeoutException('timeout for %s' % (repr(function)))

            if t._error is None:
                return t.result

        return decorator2
    return decorator



@timelimited(5)  # 设置运行超时时间2S
def fn_1(secs):
    time.sleep(secs)
    print('fn_1')
    return 'Finished without timeout'


def do_something_after_timeout():
    print('Time out!')


if __name__ == "__main__":
    try:
        print(fn_1(3))  # 设置函数执行3S
    except TimeoutException as e:
        print(str(e))
        do_something_after_timeout()

3. 装饰器（开源），不用修改原来的函数

# pip install func_timeout
https://zhuanlan.zhihu.com/p/39743129

import time
from func_timeout import func_set_timeout
from func_timeout import func_timeout, FunctionTimedOut


@func_set_timeout(2)
def task():
    while True:
        print('hello world')
        time.sleep(1)

if __name__ == '__main__':
    try:
        task()
    except FunctionTimedOut:
        print("doit('arg1', 'arg2') could not complete within 5 seconds and was terminated.
")
    except Exception as e:
        pass

4. 装饰器(os._exit(1))，不用修改原来的函数,os._exit(1)这里不好

from functools import wraps
import time
import os
from threading import Thread

def set_time_limit(t):
    def auto_quit(t1):
        '''此为控制进程超时退出的线程函数'''
        time.sleep(t1)
        print("time out {}".format(t1))
        os._exit(1) #此函数专门用于线程控制主进程退出，有兴趣的可以看一下和sys.exit()的区别
    def decorator(f):
        '''此函数用于传入被装饰函数f'''
        @wraps(f)
        def wrapper(*args,**kwargs):
            '''装饰器内部遵循的逻辑是：
            1.auto_quit先执行完，进程结束
            2.被修饰函数f先执行完，auto_quit函数停止执行
            3.被修饰函数执行完，下面的代码才能运行
            '''
            t1=Thread(target=auto_quit,args=(t,))  #此处的t是set_time_limit函数的形参，是auto_quit函数的实参
            t2=Thread(target=f,args=args,kwargs=kwargs)
            t1.setDaemon(True) #满足第2点
            t1.start()
            t2.start()
            t2.join() #满足第3点
        return wrapper
    return decorator


@set_time_limit(3)
def a(t):
    time.sleep(t)
    print("execuate finished")

if __name__=="__main__":
    a(1)
    a(5)
    a(1) # 不会打印了
    print("执行完毕")