在Python中如何在一定时间后停止函数的执行?

Question

例如，我在Python中有如下代码:

import time

def sleep(timeout: int) -> str:
time.sleep(timeout ** 7 * 1000)
return "Hello, world!"

result = sleep(5)

如您所知，这段代码非常耗时。我想这样做，如果代码的执行需要超过5秒，它将停止执行并返回一个消息，如&;function stopped&;。我该怎么做呢?

我在Python中使用了不同的东西，如时间，线程和信号，但我没有得到我想要的结果，程序崩溃了。

Answer 1

您可以使用Decorator设计模式与信号

@timeout(seconds=5, default=None)
def function():
pass
result = function()

这个timeout装饰器接受秒数作为参数，并在函数超时时返回一个默认值。然后它返回一个以目标函数为参数的装饰器函数，并返回一个带超时的调用目标函数的包装器函数。

def timeout(seconds=5, default=None):

def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
def handle_timeout(signum, frame):
raise TimeoutError()
signal.signal(signal.SIGALRM, handle_timeout)
signal.alarm(seconds)

result = func(*args, **kwargs)
signal.alarm(0)
return result
return wrapper
return decorator

Answer 2

我不太明白你说的"费时"是什么意思。但无论如何，我已经尝试过它与线程使用python在windows上，它的工作，你可以使用一个全局变量与一段时间，并管理你的函数与这样的线程:

from threading import Thread
# Your execution time in seconds
maxExecutionTime = 5
# Define a flag variable (Like a sort of On/Off switch)
global stopWorking
stopWorking = False

# Your function
def thingsToDo():
print("Function started")
while not stopWorking:
print("Function working")
# Put your code here
pass
print("Function stopped")
# Create the thread that will start your function
thread = Thread(target=thingsToDo)
# Start the thread
thread.start()
# Join your thread with the execution time you want
thread.join(maxExecutionTime)
# Set off your flag switch to indicate that the thread should stop
stopWorking = True

Answer 3

对于unix类型的系统，您可以使用SIGALRM信号。

这是我用来做这种事情的类，最后有一个用法示例。

from signal import signal, alarm, SIGALRM
import time
class TimeoutError(Exception):
...
class Timeout:
def __init__(self, seconds=1, message="Timed out"):
self._seconds = seconds
self._message = message
@property
def seconds(self):
return self._seconds
@property
def message(self):
return self._message

@property
def handler(self):
return self._handler
@handler.setter
def handler(self, handler):
self._handler = handler
def handle_timeout(self, *_):
raise TimeoutError(self.message)
def __enter__(self):
self.handler = signal(SIGALRM, self.handle_timeout)
alarm(self.seconds)
return self
def __exit__(self, *_):
alarm(0)
signal(SIGALRM, self.handler)
# example of use
with Timeout(2): # time out after two seconds with default message
try:
time.sleep(10)
except TimeoutError as e:
print(e)