Python例子

Python例子-Python设置函数最大执行时间

在我们执行python程序的时候，经常会遇到一个函数因为各种原因，程序进入了停止状态，或者卡死那里，这时候我们可以预估函数的执行时间，然后给函数设置一个最大执行时间，当超过这个执行时间的时候，函数会自动终止。python有一个func_timeout包，非常适合设置超时使用，使用起来也非常简单，代码并不多，首先得安装func_timeout。

下面举一个简单的例子:

import time
import func_timeout
from func_timeout import func_set_timeout

@func_set_timeout(5) # 设置函数最大执行时间
def test_time_out(num):
    for i in range(10):
        print(i)
        time.sleep(1)


if __name__ == '__main__':
    try:
        test_time_out(5)
    except func_timeout.exceptions.FunctionTimedOut:
        print("超时了,自动退出")

运行结果如下图所示：

从上图我们可以看出，实际上我们程序应该打印到19，但是打印到4的时候就结束了，因为我们设置了函数最大执行时间是5秒，所有打印到4的时候就结束了。

下面给大家再补充一个把函数放到线程里面的超时设置：

import time
import func_timeout
from func_timeout import func_set_timeout
import threading
from queue import Queue
@func_set_timeout(5) # 设置最大运行时间
def test_time_out(num):
    for i in range(num):
        time.sleep(1)
class DoUrl(threading.Thread):
    def __init__(self,p_name,q_list):
        threading.Thread.__init__(self, name=p_name)
        self.data = q_list
    def run(self):
        while True:
            try:
                num = self.data.get()
                try:
                    test_time_out(num)
                    print('test_time_out({})程序正常执行完成'.format(num))
                except func_timeout.exceptions.FunctionTimedOut:
                    print('test_time_out({})程序超时,退出'.format(num))
                print('还有队列{}个'.format(self.data.qsize()))
                if self.data.qsize() == 0:
                    print('线程{}完成'.format(self.getName()))
                    break
            except Exception as e:
                print(e)
                break
        return None

if __name__ == '__main__':
    q_list = Queue()
    for i in range(10):
        q_list.put(i)
    for i in range(0, 3):
        # 对象实例化
        dourl_threading = DoUrl('线程' + str(i), q_list)  # 加入队列线程
        # 启动线程
        dourl_threading.start()

输出结果如下图：

经过测试，还可以设置单个线程的运行超时时间，就是把超时装饰器放到run函数那里，代码如下：

import time
import func_timeout
from func_timeout import func_set_timeout
import threading
from queue import Queue
 # 设置最大扫描时间
def test_time_out(num):
    for i in range(num):
        time.sleep(1)
    print('test_time_out({})完成'.format(num))
class DoUrl(threading.Thread):
    def __init__(self,p_name,q_list):
        threading.Thread.__init__(self, name=p_name)
        self.data = q_list
    @func_set_timeout(100)
    def run(self):
        while True:
            try:
                num = self.data.get()
                test_time_out(num)
                print('线程{}完成处理test_time_out({})完成'.format(self.getName(),num))
                print('还有队列{}个'.format(self.data.qsize()))
                if self.data.qsize() == 0:
                    print('线程{}完成'.format(self.getName()))
                    break
            except Exception as e:
                print(e)
                break
        return None
if __name__ == '__main__':
    q_list = Queue()
    for i in range(10):
        q_list.put(i)
    for i in range(0, 3):
        # 对象实例化
        dourl_threading = DoUrl('线程' + str(i), q_list)  # 加入队列线程
        # 启动线程
        dourl_threading.start()

当我把线程超时时间设置为100的时候，线程正常完成，如下图：

但是当我把run函数的超时时间设置成10秒的时候，就报错，然后退出了线程，如下图：

虽然报错了，但是10秒里面程序还是执行了的啊

最后修改:2021-07-07 01:33:33