eventlet

import time
import eventlet  #导入eventlet这个模块
eventlet.monkey_patch()   #必须加这条代码
with eventlet.Timeout(2,False):   #设置超时时间为2秒
   print '这条语句正常执行'
   time.sleep(4)
   print '没有跳过这条输出'
print '跳过了输出'

问题:
1、针对子进程无法跳出

timeout_decorator

from tqdm import trange
import sys
import time
import timeout_decorator

@timeout_decorator.timeout(int(sys.argv[2]))
def test():
    if sys.argv[1] == '--timeout':
        for i in trange(3):
            time.sleep(1)
            print ('>>> {} seconds passed.'.format(i+1))
    return 0

if __name__ == '__main__':
    try:
        test()
    except Exception as e:
        print ('Timeout Error Catched!')
        print (e)
    print ("Timeout Task Ended!")

问题:
1、报错signal only works in main thread
在默认情况下,timeout-decorator使用信号来限制给定函数的执行时间。

如果您的函数不在主线程中执行(例如,如果它是Web应用程序的工作线程),则此配置不起作用。在这种情况下使用多进程来作为超时替代策略。要使用它,只需将use_signals = False传递给超时装饰器函数:

import time
import timeout_decorator

@timeout_decorator.timeout(5, use_signals=False)
def mytest():
    print "Start"
    for i in range(1,10):
        time.sleep(1)
        print("{} seconds have passed".format(i))

if __name__ == '__main__':
    mytest()

2、多线程使用会有问题,在多处理策略超时的情况下,您的函数不会返回无法被pickle的对象,否则它将在主进程和子进程之间进行编组时失败。

signal


import signal

class TimeOutException(Exception):
    pass


def handle(signum, frame):
    raise TimeOutException("运行超时!")


def set_timeout(timeout, callback):
    def wrapper(func):
        def inner(*args, **kwargs):
            try:
                signal.signal(signal.SIGALRM, handle)
                signal.alarm(timeout)  # 开启闹钟信号
                rs = func(*args, **kwargs)
                signal.alarm(0)  # 关闭闹钟信号
                return rs
            except TimeOutException as e:
                callback()

        return inner

    return wrapper


def process_time():
    print("超时了")


@set_timeout(40, process_time)
def function():
	pass
	

超时信号只能设置在主线程函数,不能设置在子线程函数

Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐