python程序超时处理方式
作者：何小义的AI进阶路
这篇文章主要介绍了python程序超时处理方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
python程序超时处理
下面这个小案例，实现对某个函数超时处理：
其实就是加了一个装饰器timeout
import time
import timeout_decorator
@timeout_decorator.timeout(6)
def justtest():
time.sleep(5)
return 5
aaa = justtest()
print aaa上面代码我们规定若函数运行时间超过6s 就报错。
5s是可以的，所以正常运行，运行结果：
如果我们设定程序不可超过3s，
import time
import timeout_decorator
@timeout_decorator.timeout(3)
def justtest():
time.sleep(5)
return 5
aaa = justtest()
print aaa运行结果：
python超时处理方法介绍，Eventlet 和 func-timeout
有一些场景，需要我们控制程序或者脚本的运行时间。
- 自动化用例中的某一个步骤需要在规定时间内完成才算有效；
- 使用线程池控制线程运行时，我们不希望每个线程运行时间超过某一值；
- 在爬虫时发送的request请求，我们希望在规定时间内无响应则跳过该请求；
- 当写入了一条错误的sql查询语句可能会导致该查询语句一直运行，导致其他查询语句阻塞时等等场景
以上场景中，如果我们在程序中设置了超时处理，那将会为我们的程序节约很多运行时间。
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：
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',)) (kwargs={}) timed out # after 3.000000 seconds.
# 进程运行超时
# 说明：func_timeout()第一个参数为超时时间，第二个参数是调用的函数，注意这里只调用函数名，没有(), 第三个参数是调用函数的参数。这里使用的异常捕获也不能解决2.1中的报错问题。总结
以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。