python防止程序超时的实现示例

# func_set_timeout作为装饰器使用，来作为目标函数的超时处理方式
import time
import os
from func_timeout import func_set_timeout
@func_set_timeout(5)
def my_test(name):
    print('子进程运行中，name={}，pid={}'.format(name, os.getpid()))
    time.sleep(4)			
    print('子进程已经结束')
if __name__ == '__main__':
    print('父进程：{}'.format(os.getpid()))
    try:
        p = Process(target=my_test, args=('test', ))
        p.start()
    except TimeoutError as e:
        print('子程序超时')

import time
import threading
def worker():
    print('worker start')
    time.sleep(10)
    print('worker end')
t = threading.Thread(target=worker)
t.start()
# 等待线程结束，不能超过5秒
t.join(5)
if t.is_alive():
    print('worker overtime')
else:
    print('worker finished')

"""
用于处理函数超时的情况,同时记录程序异常
"""
import eventlet
class PyTimer:
    """
    用于在处理异常的同时防止程序超时
    """
    def __init__(self, max_time=5):
        """
        初始化
        """
        # 必须有这条代码
        eventlet.monkey_patch()
        self.errors = ""
        # 最大超时的秒数
        self.max_time = max_time
    def set_max_time(self,new_time):
        """
        设置新的超时秒数
        """
        self.max_time=new_time
    def count_time(self, function, *args):
        """
        运行函数，记录异常，同时超时跳出
        """
        return_result = None
        timeout = True
        current_error = ""
        try:
            with eventlet.Timeout(self.max_time, False):  # 设置超时时间为5秒
                return_result = function(*args)
                timeout = False
        except Exception as e:
            current_error = str(e) + "\n"
        finally:
            if timeout:
                current_error += function.__name__ + "超时\n"
                print(current_error)
                self.errors += current_error
            return return_result
    def get_errors(self):
        """
        获取报错信息
        """
        return self.errors
if __name__ == '__main__':
    def long_fun(a, b):
        import time
        print(a, b)
        time.sleep(20)
    my_timer = PyTimer()
    my_timer.count_time(long_fun, 1, 2)