在一定时间后中断该功能

问题描述：

在 Python 中，举一个玩具示例：

for x in range(0, 3):
    # Call function A(x)

for如果函数 A 花费的时间超过五秒，我想通过跳过它来继续循环，这样我就不会陷入困境或浪费时间。

通过一些搜索，我意识到子进程或线程可能会有帮助，但我不知道如何在这里实现它。

解决方案 1：

我认为创建新进程可能有点过头了。如果您使用的是 Mac 或基于 Unix 的系统，则应该能够使用 signal.SIGALRM 强制使耗时过长的函数超时。这将适用于因网络或其他问题而空闲的函数，而您绝对无法通过修改函数来处理这些问题。我在这个答案中有一个使用它的示例：

SSH 短时间后超时的选项？ClientAlive 和 ConnectTimeout 似乎无法满足我的要求

在这里编辑我的答案，但我不确定是否应该这样做：

import signal

class TimeoutException(Exception):   # Custom exception class
    pass

def timeout_handler(signum, frame):   # Custom signal handler
    raise TimeoutException

# Change the behavior of SIGALRM
signal.signal(signal.SIGALRM, timeout_handler)

for i in range(3):
    # Start the timer. Once 5 seconds are over, a SIGALRM signal is sent.
    signal.alarm(5)    
    # This try/except loop ensures that 
    #   you'll catch TimeoutException when it's sent.
    try:
        A(i) # Whatever your function that might hang
    except TimeoutException:
        continue # continue the for loop if function A takes more than 5 second
    else:
        # Reset the alarm
        signal.alarm(0)

这基本上会设置一个 5 秒的计时器，然后尝试执行您的代码。如果在时间用完之前无法完成，则会发送 SIGALRM，我们会捕获该 SIGALRM 并将其转换为 TimeoutException。这会强制您进入 except 块，您的程序可以继续。

解决方案 2：

根据 TheSoundDefense 的回答，也许有人认为这个装饰器很有用：

import time
import signal

class TimeoutException(Exception):   # Custom exception class
    pass


def break_after(seconds=2):
    def timeout_handler(signum, frame):   # Custom signal handler
        raise TimeoutException
    def function(function):
        def wrapper(*args, **kwargs):
            signal.signal(signal.SIGALRM, timeout_handler)
            signal.alarm(seconds)
            try:
                res = function(*args, **kwargs)
                signal.alarm(0)      # Clear alarm
                return res
            except TimeoutException:
                print u'Oops, timeout: %s sec reached.' % seconds, function.__name__, args, kwargs
            return
        return wrapper
    return function

测试：

@break_after(3)
def test(a, b, c):
    return time.sleep(10)

>>> test(1,2,3)
Oops, timeout: 3 sec reached. test (1, 2, 3) {}

解决方案 3：

如果您可以将您的工作分开并时不时地检查，这几乎总是最好的解决方案。但有时这是不可能的——例如，也许您正在从一个缓慢的文件共享中读取一个文件，而这个文件共享偶尔会挂起 30 秒。要从内部处理这个问题，您必须围绕异步 I/O 循环重构整个程序。

如果不需要跨平台，你可以在*nix（包括Mac和Linux）上使用信号，在Windows上使用APC等。但如果您需要跨平台，那就行不通了。

因此，如果您确实需要同时执行此操作，那么您可以这样做，有时您不得不这样做。在这种情况下，您可能希望为此使用进程，而不是线程。您无法真正安全地终止线程，但您可以终止进程，并且可以像您希望的那样安全。此外，如果线程由于受 CPU 限制而花费 5 秒以上的时间，您不想与它争夺 GIL。

这里有两个基本选项。

首先，您可以将代码放入另一个脚本中并使用以下命令运行它subprocess：

subprocess.check_call([sys.executable, 'other_script.py', arg, other_arg],
                      timeout=5)

由于这是通过正常的子进程通道进行的，因此您可以使用的唯一通信是一些argv字符串、成功/失败的返回值（实际上是一个小的整数，但这并不是很好），以及可选的输入一大块文本和输出一大块文本。

或者，您可以使用multiprocessing来生成类似线程的子进程：

p = multiprocessing.Process(func, args)
p.start()
p.join(5)
if p.is_alive():
    p.terminate()

如您所见，这有点复杂，但在某些方面更好：

• 您可以传递任意 Python 对象（至少是任何可以被 pickle 的对象），而不仅仅是字符串。

• 您不必将目标代码放在完全独立的脚本中，而是可以将其作为函数保留在同一个脚本中。

• 它更加灵活 - 例如，如果您以后需要传递进度更新，则可以很容易地在任一方向或两个方向添加队列。

任何类型的并行性都存在一个大问题，那就是共享可变数据——例如，让后台任务更新全局字典作为其工作的一部分（您的评论表明您正在尝试这样做）。使用线程，您可以侥幸逃脱，但竞争条件可能会导致数据损坏，因此您必须非常小心地锁定。对于子进程，您根本无法逃脱。（是的，您可以使用共享内存，正如进程间共享状态所解释的那样，但这仅限于简单类型，如数字、固定数组和您知道如何定义为 C 结构的类型，它只会让您回到与线程相同的问题。）

理想情况下，您可以安排好一切，这样在进程运行时就不需要共享任何数据 — 您传入一个dict参数，然后返回dict一个结果。当您有一个想要放在后台的先前同步的函数时，这通常很容易安排。

但是，如果说部分结果比没有结果要好呢？在这种情况下，最简单的解决方案是通过队列传递结果。您可以使用显式队列来执行此操作，如在进程之间交换对象中所述，但还有一种更简单的方法。

如果你可以将整体流程分解为单独的任务，每个任务对应一个你想保留在字典中的值（或一组值），那么你可以将它们安排在Pool— 或者更好的是 上concurrent.futures.Executor。（如果你使用的是 Python 2.x 或 3.1，请参阅 PyPI 上的反向移植futures。）

假设你的慢速函数如下所示：

def spam():
    global d
    for meat in get_all_meats():
        count = get_meat_count(meat)
        d.setdefault(meat, 0) += count

相反，你应该这样做：

def spam_one(meat):
    count = get_meat_count(meat)
    return meat, count

with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:
    results = executor.map(spam_one, get_canned_meats(), timeout=5)
    for (meat, count) in results:
        d.setdefault(meat, 0) += count

您在 5 秒内获得的结果都会被添加到字典中；如果不是全部，则其余的结果将被放弃，并TimeoutError引发（您可以随意处理 - 记录它，执行一些快速回退代码，等等）。

如果任务确实是独立的（就像我这个愚蠢的小例子一样，但当然它们可能不存在于你的真实代码中，至少在没有进行重大重新设计的情况下），你只需删除它就可以免费并行化工作max_workers=1。然后，如果你在一台 8 核机器上运行它，它会启动 8 个工作者并给他们每个人 1/8 的工作，事情就会完成得更快。（通常不是 8 倍快，但通常是 3-6 倍快，这仍然相当不错。）

解决方案 4：

这似乎是一个更好的主意（抱歉，我还不确定该事物的 Python 名称）：

import signal

def signal_handler(signum, frame):
    raise Exception("Timeout!")

signal.signal(signal.SIGALRM, signal_handler)
signal.alarm(3) # Three seconds
try:
    for x in range(0, 3):
        # Call function A(x)
except Exception, msg:
    print "Timeout!"
signal.alarm(0) # Reset

解决方案 5：

注释是正确的，您应该检查内部。这是一个潜在的解决方案。请注意，异步函数（例如使用线程）与此解决方案不同。这是同步的，这意味着它仍将以串行方式运行。

import time

for x in range(0,3):
    someFunction()

def someFunction():
    start = time.time()
    while (time.time() - start < 5):
        # do your normal function

    return;