Python中如何在任务超时（超5秒）时切换执行其他任务？

嘿，这个需求其实挺常见的，在Python里我们有几种可靠的方式来实现，我给你推荐两种最实用的方案，按需选择就行：

方案一：用concurrent.futures.ThreadPoolExecutor（通用推荐） #

这个方法兼容性好，不管是IO密集型还是大多数场景都适用，而且不需要修改你原有的function(a)代码。核心思路是把耗时任务放到子线程里执行，主线程等待指定时间，超时就触发分支逻辑。

from concurrent.futures import ThreadPoolExecutor, TimeoutError

def function(a):
    # 你的耗时函数逻辑，比如模拟长时间运行
    import time
    time.sleep(6)  # 这里改成你实际的代码
    return "function(a)完成"

def function(b):
    print("执行function(b)：因为function(a)超时了")

def function(c):
    print("执行function(c)：因为function(a)没超时")

def main():
    a_param = "你的参数a"
    with ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(function, a_param)
        try:
            # 等待5秒，获取执行结果
            result = future.result(timeout=5)
            print(f"function(a)执行结果：{result}")
            function(c)
        except TimeoutError:
            print("function(a)执行超过5秒，终止任务（注：子线程可能仍在后台运行）")
            function(b)

if __name__ == "__main__":
    main()

注意点： #

• 这里的“终止任务”其实是主线程不再等待子线程，但子线程本身可能会继续运行直到完成。如果需要真正强制终止子线程，你得修改function(a)，让它支持中断（比如定期检查一个标志位），示例如下：

import threading

def function(a, stop_event):
    import time
    for i in range(10):
        if stop_event.is_set():
            print("function(a)被强制终止")
            return
        time.sleep(1)
    return "function(a)完成"

def main():
    a_param = "你的参数a"
    stop_event = threading.Event()
    with ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(function, a_param, stop_event)
        try:
            result = future.result(timeout=5)
            print(f"function(a)执行结果：{result}")
            function(c)
        except TimeoutError:
            print("function(a)超时，强制终止")
            stop_event.set()
            function(b)

方案二：用signal模块（仅适用于主线程+类Unix系统） #

如果你的代码是在类Unix系统（Linux/macOS）的主线程中运行，也可以用信号机制来实现超时控制，这种方式能真正中断函数执行，但局限性比较大：

import signal

class TimeoutError(Exception):
    pass

def timeout_handler(signum, frame):
    raise TimeoutError("执行超时")

def function(a):
    import time
    time.sleep(6)
    return "function(a)完成"

def function(b):
    print("执行function(b)：因为function(a)超时了")

def function(c):
    print("执行function(c)：因为function(a)没超时")

def main():
    a_param = "你的参数a"
    # 设置信号处理器，5秒后触发SIGALRM
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(5)
    try:
        result = function(a_param)
        signal.alarm(0)  # 取消未触发的闹钟
        print(f"function(a)执行结果：{result}")
        function(c)
    except TimeoutError:
        print("function(a)执行超过5秒")
        function(b)

if __name__ == "__main__":
    main()

局限性： #

• 只能在主线程中使用，子线程里无法通过signal设置超时
• Windows系统不支持SIGALRM信号，所以这个方案在Windows上无效

内容的提问来源于stack exchange，提问作者Mahek Shah