Python 中限制函数执行

类似的问题和答案还有很多，但我仍然找不到可靠的答案。

所以，我有一个函数，可能运行时间太长。函数是私有的，从某种意义上说我无法更改它的代码。

我想将其执行时间限制为 60 秒。 我尝试了以下方法：

1. Python 信号。不适用于 Windows 和多线程环境 (mod_wsgi)。
2. 线程。好方法，但是线程无法停止，因此即使在提升后它仍然存在TimeoutException.
3. multiprocessing蟒蛇模块。我在腌制方面遇到问题，我不知道如何解决。我想制作 time_limit 装饰器，但在顶层导入所需函数时存在问题。执行时间较长的函数是实例方法，包装它也没有帮助......

那么，对于上述问题有没有好的解决方案呢？ 如何杀死我启动的线程？ 如何使用子流程并避免酸洗问题？ 是subprocess模块有什么帮助吗？

谢谢。

我认为多处理方法是您唯一真正的选择。您是对的，线程不能被杀死（很好）并且信号存在跨平台问题。这是一种多处理实现：

import multiprocessing
import Queue

def timed_function(return_queue):
    do_other_stuff()
    return_queue.put(True)
    return

def main():

    return_queue = multiprocessing.Manager().Queue()

    proc = multiprocessing.Process(target=timed_function, args=(return_queue,))
    proc.start()

    try:

        # wait for 60 seconds for the function to return a value
        return_queue.get(timeout=60)

    except Queue.Empty:
        # timeout expired
        proc.terminate() # kill the subprocess
        # other cleanup

我知道你说过你有酸洗问题，但这些问题几乎总是可以通过重构来解决。例如，你说你的长函数是一个实例方法。你can包装这些类型的函数以将它们用于多处理：

class TestClass(object):
    def timed_method(self, return_queue):
        do_other_stuff()
        return_queue.put(True)
        return

要在工作人员池中使用该方法，请将此包装器添加到模块的顶层：

def _timed_method_wrapper(TestClass_object, return_queue):
    return TestClass_object(return_queue)

例如，现在您可以使用apply_async来自同一类的不同方法的此类方法：

def run_timed_method():
    return_queue = multiprocessing.Manager().Queue()
    pool = multiprocessing.Pool()
    result = pool.apply_async(_timed_method_wrapper, args=(self, return_queue))

我很确定只有当您使用多处理池而不是使用启动子进程多处理进程目的。另外，我敢打赌很多人会对这种构造皱眉，因为你破坏了类提供的漂亮、干净的抽象，并且还在类和其他随机包装函数之间创建了依赖关系。你必须自己决定让你的代码变得更丑是否值得。