timeout限制运行时间

在编程中，限制代码运行时间是一个非常重要的功能。这不仅可以防止程序因死循环或长时间运行而耗尽资源，还可以提高系统的稳定性和响应速度。下面我们将详细介绍如何在Python中使用timeout来限制代码的运行时间，并提供具体的实践步骤和代码示例。

1. 使用信号模块实现超时控制

在Unix/Linux系统上，我们可以使用Python的signal模块来设置一个定时器，如果代码执行超过指定的时间，就会触发一个超时异常。

实践步骤：

1. 导入必要的模块：signal 和 time。
2. 定义一个处理超时的函数。
3. 使用signal.alarm()设置定时器。
4. 捕获超时异常并处理。

示例代码：

import signal
import time

class TimeoutException(Exception):
    pass

def timeout_handler(signum, frame):
    raise TimeoutException("Time out!")

def run_with_timeout(func, args=(), kwargs={}, timeout_duration=1):
    # 设置信号处理函数
    signal.signal(signal.SIGALRM, timeout_handler)
    # 设置计时器
    signal.alarm(timeout_duration)
    try:
        result = func(*args, **kwargs)
        # 如果函数正常结束，取消计时器
        signal.alarm(0)
        return result
    except TimeoutException as te:
        print(te)
    finally:
        # 确保最后清除信号处理函数
        signal.signal(signal.SIGALRM, signal.SIG_IGN)

# 测试函数
def long_running_function():
    try:
        print("Start sleeping...")
        time.sleep(5)  # 模拟长时间运行的任务
        print("End sleeping")
    except Exception as e:
        print(e)

# 运行测试
if __name__ == "__main__":
    run_with_timeout(long_running_function, timeout_duration=2)

2. 使用多线程实现超时控制

在Windows系统上，由于signal模块的功能有限，我们可以通过创建一个新线程来实现超时控制。

实践步骤：

1. 导入threading模块。
2. 创建一个新的线程来执行目标函数。
3. 使用join()方法等待线程完成，并设置超时时间。
4. 如果线程在指定时间内未完成，则终止线程。

示例代码：

import threading
import time

class TimeoutThread(threading.Thread):
    def __init__(self, func, args=(), kwargs={}):
        super().__init__()
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self.result = None
        self.exception = None

    def run(self):
        try:
            self.result = self.func(*self.args, **self.kwargs)
        except Exception as e:
            self.exception = e

def run_with_timeout_thread(func, args=(), kwargs={}, timeout_duration=1):
    thread = TimeoutThread(func, args, kwargs)
    thread.start()
    thread.join(timeout_duration)
    if thread.is_alive():
        print("Function timed out!")
        return None
    if thread.exception:
        raise thread.exception
    return thread.result

# 测试函数
def long_running_function():
    try:
        print("Start sleeping...")
        time.sleep(5)  # 模拟长时间运行的任务
        print("End sleeping")
        return "Done"
    except Exception as e:
        print(e)

# 运行测试
if __name__ == "__main__":
    result = run_with_timeout_thread(long_running_function, timeout_duration=2)
    print("Result:", result)