实际项目中会涉及到需要对有些函数的响应时间做一些限制，如果超时就退出函数的执行，停止等待。
可以利用python中的装饰器实现对函数执行时间的控制。
python装饰器简单来说可以在不改变某个函数内部实现和原来调用方式的前提下对该函数增加一些附件的功能，提供了对该函数功能的扩展。
方法一. 使用 signal
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
# 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
def after_timeout(): # 超时后的处理函数
print("time out!")
@set_timeout(2, after_timeout) # 限时 2 秒超时
def connect(): # 要执行的函数
time.sleep(3) # 函数执行时间，写大于2的值，可测试超时
print('finished without timeout.')
if __name__ == '__main__':
connect()
|
方法一中使用的signal有所限制，需要在linux系统上，并且需要在主线程中使用。方法二使用线程计时，不受此限制。
方法二. 使用thread
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
|
# -*- coding: utf-8 -*-
from threading import thread
import time
class timeoutexception(exception):
pass
threadstop = thread._thread__stop
def timelimited(timeout):
def decorator(function):
def decorator2(*args,**kwargs):
class timelimited(thread):
def __init__(self,_error= none,):
thread.__init__(self)
self._error = _error
def run(self):
try:
self.result = function(*args,**kwargs)
except exception,e:
self._error = str(e)
def _stop(self):
if self.isalive():
threadstop(self)
t = timelimited()
t.start()
t.join(timeout)
if isinstance(t._error,timeoutexception):
t._stop()
raise timeoutexception('timeout for %s' % (repr(function)))
if t.isalive():
t._stop()
raise timeoutexception('timeout for %s' % (repr(function)))
if t._error is none:
return t.result
return decorator2
return decorator
@timelimited(2) # 设置运行超时时间2s
def fn_1(secs):
time.sleep(secs)
return 'finished without timeout'
def do_something_after_timeout():
print('time out!')
if __name__ == "__main__":
try:
print(fn_1(3)) # 设置函数执行3s
except timeoutexception as e:
print(str(e))
do_something_after_timeout()
|
总结
以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，谢谢大家对服务器之家的支持。如果你想了解更多相关内容请查看下面相关链接
原文链接：https://blog.csdn.net/dcrmg/article/details/82850457