有什么好办法约束一个函数的执行时间吗?

创建主题时描述错了,为看到的 v 友说声抱歉, Process 和 Thread 说的乱七八糟的, 已改正.

docker

celery 🙈

老哥，找到什么好方法了公布下

创建一个进程之后获取到 pid，时间到了直接 kill

你是要开发实时系统吗？楼主你说的都是软实时的，简单来说，内核对你的进程进行调度，那么你的函数执行时间上限就有可能会超过了。真要硬实时的话我还是建议你用 C，然后老实地独占一个核心。

不知道是不是想要这种：把函数内的执行代码分成小块，然后每一小块执行完后检查执行时间，超时既返回。如果执行自己写的函数，可以用这种方法。

@HarveyDent 比如有一句 sleep(10), 这句话没法再被拆分了.

```
def run_with_limited_time(func, args, kwargs, time):
"""Runs a function with time limit
:param func: The function to run
:param args: The functions args, given as tuple
:param kwargs: The functions keywords, given as dict
:param time: The time limit in seconds
:return: True if the function ended successfully. False if it was terminated.
"""
def wrapper(queue, func, *args, **kwargs):
return queue.put(func(*args, **kwargs))
import multiprocessing
q = multiprocessing.Queue(maxsize=1)
func = functools.partial(wrapper, q, func)
p = Process(target=func, args=args, kwargs=kwargs)
p.start()
p.join(time)
if p.is_alive():
p.terminate()
raise TimeoutError('time out!')

return q.get_nowait()
```

看起来是最好的实现了

循环里设置标记位置,循环外部能控制这个标记位置让循环退出

@lolizeppelin 不行。

sleep 可以同信号抛出异常的

哦 重新看了下你说的

用 eventlet 写协程可以做到
因为里面没有真正的 sleep
sleep 都被 hack 成排序调度的位置了

协程本身可以终止

# 超时装饰器
import signal,functools # 下面会用到的两个库
class TimeoutError(Exception): pass # 定义一个 Exception，后面超时抛出

def timeout(seconds, error_message = 'Function call timed out'):
def decorated(func):
def _handle_timeout(signum, frame):
raise TimeoutError(error_message)
def wrapper(*args, **kwargs):
signal.signal(signal.SIGALRM, _handle_timeout)
signal.alarm(seconds)
try:
result = func(*args, **kwargs)
finally:
signal.alarm(0)
return result
return functools.wraps(func)(wrapper)
return decorated