python 限制函数执行时间 代码实现

# coding=utf-8
import signal
import time


def set_timeout(num, callback):
    def wrap(func):
        def handle(signum, frame):  # 收到信号 SIGALRM 后的回调函数，第一个参数是信号的数字，第二个参数是the interrupted stack frame.
            raise RuntimeError

        def to_do(*args, **kwargs):
            try:
                signal.signal(signal.SIGALRM, handle)  # 设置信号和回调函数
                signal.alarm(num)  # 设置 num 秒的闹钟
                print('start alarm signal.')
                r = func(*args, **kwargs)
                print('close alarm signal.')
                signal.alarm(0)  # 关闭闹钟
                return r
            except RuntimeError as e:
                callback()

        return to_do

    return wrap


if __name__ == '__main__':
    def after_timeout():  # 超时后的处理函数
        print("do something after timeout.")


    @set_timeout(2, after_timeout)  # 限时 2 秒
    def connect():  # 要执行的函数
        time.sleep(1)  # 函数执行时间，写大于2的值，可测试超时
        return 'connect success.'


    print(connect())

注意事项：
　　

　　电脑系统是win10 64位，在使用python的signal模块时报错：“AttributeError: module 'signal' has no attribute 'SIGALRM'”，这是由于signal模块能够在linux下正常使用，但在windows下却有一些限制，在python文档https://docs.python.org/2/library/signal.html#signal.signal找到了以下解释：

"On Windows, signal() can only be called with SIGABRT, SIGFPE, SIGILL, SIGINT, SIGSEGV, or SIGTERM. A ValueError will be raised in any other case."

也就是说在windows下只能使用这几个信号：

　　SIGABRT
　　SIGFPE
　　SIGILL
　　SIGINT
　　SIGSEGV
　　SIGTERM

 

也可经过线程的方式 设置超时 ，主线程设置超时时间 ， 杀死子线程

import timefrom threading import Threadimport inspectimport ctypesdef _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 to_do():    print('子线程开始了')    time.sleep(50)    print('结束了')if __name__ == '__main__':    t = Thread(target=to_do)    t.start()    stop_thread(t)    print('主线程结束')下面这种方案 ，缺陷是，设置好超时后，子线程结束，主线程也不会中止，有可能会下降效率能够将stop_thread(t)  换为    t.join( num  秒 ）  来控制  超时kill