无需使用contextlib实现函数超时，咨询非主线程可行方案

由于signal模块仅支持主线程接收信号，非主线程中需改用线程调度+超时检测的方案，以下提供两种无contextlib依赖的实用实现：

方案一：基于concurrent.futures.ThreadPoolExecutor（推荐） #

利用线程池的任务提交机制，通过result(timeout)直接设置超时阈值，超时会触发内置的TimeoutError，无需额外依赖，且可在非主线程中正常运行。

示例代码：

import concurrent.futures
import time

# 自定义超时异常（可选，也可直接使用库内置TimeoutError）
class FunctionTimeout(Exception):
    pass

def long_function():
    time.sleep(15)
    return "任务执行完成"

def run_with_timeout(func, timeout_seconds, *args, **kwargs):
    # 初始化单线程池（仅用于执行目标函数）
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
        future_task = executor.submit(func, *args, **kwargs)
        try:
            # 等待任务结果，超时则抛出异常
            return future_task.result(timeout=timeout_seconds)
        except concurrent.futures.TimeoutError:
            raise FunctionTimeout(f"函数执行超时，已超过{timeout_seconds}秒")

# 测试：在非主线程中调用超时逻辑
import threading
def non_main_thread_task():
    print("非主线程启动")
    try:
        result = run_with_timeout(long_function, 5)
        print(result)
    except FunctionTimeout as e:
        print(e)

# 启动非主线程
t = threading.Thread(target=non_main_thread_task)
t.start()
t.join()

方案二：基于threading.Timer + ctypes底层实现 #

若不想依赖线程池，可通过定时线程触发，结合ctypes向目标线程强制抛出异常（注：此为底层操作，需注意线程安全）。

示例代码：

import threading
import time
import ctypes

class FunctionTimeout(Exception):
    pass

def raise_exception_in_thread(thread_id, exc_type):
    # 通过ctypes向指定线程异步抛出异常
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(
        ctypes.c_long(thread_id),
        ctypes.py_object(exc_type)
    )
    if res == 0:
        raise ValueError("无效的线程ID")
    elif res > 1:
        # 异常设置失败时恢复线程状态
        ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(thread_id), None)
        raise SystemError("异步异常设置失败")

def run_with_timeout(func, timeout_seconds, *args, **kwargs):
    result = None
    task_exc = None
    def task_target():
        nonlocal result, task_exc
        try:
            result = func(*args, **kwargs)
        except Exception as e:
            task_exc = e

    # 启动目标线程执行函数
    task_thread = threading.Thread(target=task_target)
    task_thread.start()
    # 启动超时定时器，到期后向目标线程抛异常
    timeout_timer = threading.Timer(timeout_seconds, raise_exception_in_thread, args=(task_thread.ident, FunctionTimeout))
    timeout_timer.start()
    # 等待目标线程执行结束
    task_thread.join()
    # 若任务提前完成，取消定时器
    timeout_timer.cancel()
    # 传递执行结果或异常
    if task_exc:
        raise task_exc
    return result

# 测试非主线程调用
def non_main_thread_task():
    print("非主线程启动")
    try:
        run_with_timeout(long_function, 5)
    except FunctionTimeout as e:
        print(f"超时触发：{e}")

t = threading.Thread(target=non_main_thread_task)
t.start()
t.join()

注意事项 #

• 方案一线程池方式更简洁安全，无底层兼容性问题，优先推荐；
• 方案二的ctypes方式属于底层操作，可能在部分Python版本/环境下存在兼容性风险，需谨慎使用；
• 无论哪种方案，超时后目标函数的线程会被终止或抛出异常，若函数涉及资源操作，需在函数内部通过try/finally确保资源释放。

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