python程序超时处理方式

2023-11-19 10:49 开发作者：何小义的AI进阶路

python程序超时处理

下rFPrtd面这个小案例，实现对某个函数超时处理：

其实就是加了一个装饰器timeout

import time
import timeout_decorator
@timeout_dechttp://www.devze.comorator.timeout(6)
def justtest():
    time.sleep(5)
    return 5
aaa = justtest()
print aaa

上面代码我们规定若函数运行时间超过6s 就报错。

5s是可以的，所以正常运行，运行结果：

python程序超时处理方式

如果我们设定程序不可超过3s，

import time
import timeout_decorator
@timeout_decorator.timeout(3)
def justtest():
    time.sleep(5)
    return 5
aaa = justtest()
print aaa

运行结果：

python程序超时处理方式

python超时处理方法介绍，Eventlet 和 func-timeout

有一些场景，需要我们控制程序或者脚本的运行时间。

自动化用例中的某一个步骤需要在规定时间内完成才算有效； hpp;
使用线程池控制线程运行时，我们不希望每个线程运行时间超过某一值；
在爬虫时发送的request请求，我们希望在规定时间内无响应则跳过该请求；
当写入了一条错误的sql查询语句可能会导致该查询语句一直js运行，导致其他查询语句阻塞时等等场景

以上场景中，如果我们在程序中设置了超时处理，那将会为我们的程序节约很多运行时间。

1.eventlet

安装：

pip install eventlet

eventlet在代码中的使用：

import time
import eventlet                     # 导入eventlet模块
eventlet.monkey_patch()             # 引入patch
with eventlet.Timeout(3, False):    # 设置超时时间3秒
    time.sleep(4)                   # 等待4秒
    print('结果1')                   # 因为等待了4秒，大于超时时间，所以'结果1‘不打印
print('结果2')
# 运行以上程序输出：
# 结果2
with eventlet.Timeout(3, False):    # 设置超时时间3秒
    time.sleep(2)                   # 等待2秒
    print('结果3')                   # 因为等待了2秒，小于超时时间，所以'结果3‘也会打印
print('结果4')
# 运行以上程序输出：
# 结果3
# 结果4

注意：eventlet方法中只要进行了子进程的调用，超时设置就会失效。示例：

import time
import os
import eventlet                     # 导入eventlet模块
from multiprocessing import Process
def my_test(name):
    print('子进程运行中，name={}，pid={}'.format(name, os.getpid()))
    time.sleep(10)
    print('子进程已经结束')
if __name__ == '__main__':
    eventlet.monkey_patch()                                 # 引入patch
    with eventlet.Timeout(3, False):                        # 这里设置了超时时间3
        print('父进程：{}'.format(os.getpid()))
        p = Process(target=my_test, args=('test', ))        # 调用了子进程，子进程内有等待10秒的步骤，等待完才会打印'子进程已经结束‘
        print('子进程准备执行---')
        p.start()
# 执行以上程序输出：
# 父进程：11548
# 子进程准备执行 - --
# 子进程运行中，name = test，pid = 1996
# 子进程已经结束

从以上示例可以看出，设置的超时时间3秒没有生效。这种情况，我们只能将超时处理步骤加到my_test函数中即可：

def my_test(name):
    print('子进程运行中，name={}，pid={}'.format(name, os.getpid()))
    eventlet.monkey_patch()                                 # 引入patch
    with eventlet.Timeout(3, False):
        time.sleep(10)
        print('等待时间结束')
    print('子进程已经结束')

2.func-timeout

安装：

pip install func-timeout

func-timeout 在代码中的使用，func-timeout主要有两中方法：func_timeout，func_set_timeout

2.1 func_set_timeout 用法

# 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('子程序超时')
# 执行以上程序输出：
# 父进程：29908
# 结束
# 子进程运行中，name=test，pid=40280
# 子进程已经结束
# 说明：my_test函数执行需要4s，设置的超时时间大于4s，因此子进程正常执行；若设置my_test函数执行时间大于5s，则报错，代码中尝试了通过捕获异常来让程序正常运行退出，但超时异常时发生在子进程中的，因此捕获异常失败，此时的报错见图1。

图1：

python程序超时处理方式

2.2 func_timeout 用法

import time
import os
from func_timeout import func_timeout
from func_timeout import FunctionTimedOut
def my_test(name):
    print('进程运行中，name={}，pid={}'.format(name, os.getpid()))
    time.sleep(6)
    print('进程已经结束')
if __name__ == '__main__':
    try:
        func_timeout(3, my_test, args=('test', ))
    except FunctionTimedOut as e:
        print(e)
        print('进程运行超时')
# 输出：
# 进程运行中，name=test，pid=3364
# Function my_test (args=('test',)) (kwarg编程s={}) timed out # after 3.000000 seconds.
# 进程运行超时
# 说明：func_timeout()第一个参数为超时时间，第二个参数是调用的函数，注意这里只调用函数名，没有(), 第三个参数是调用函数的参数。这里使用的异常捕获也不能解决2.1中的报错问题。

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持编程客栈(www.devze.com)。

继续阅读：python程序超时 python程序超时处理 python超时处理

python程序超时处理方式

目录