Python中抛出异常后函数未停止运行的问题求助

你遇到的核心问题是**threading.Timer在子线程抛出的异常无法中断主线程中目标函数的执行**，而且Python没有安全的方式强制终止线程——线程一旦启动，只能等待其自行结束。你当前的实现里，Timer只是在超时后抛异常，但主线程里的test()函数仍会继续运行直到完成。

解决方案：改用进程实现超时终止 #

改用multiprocessing模块创建独立进程来运行目标函数，进程支持terminate()方法可以强制终止，完美解决超时终止的需求。同时用psutil监控子进程的内存占用，用时间戳计算执行时长。

修改后的完整代码 #

import time
import psutil
import multiprocessing
from multiprocessing import Queue

class TimeoutError(Exception):
    pass

class Performance:
    def __init__(self, func, time_limit=20, except_return=None) -> None:
        self.func = func
        self.except_return = except_return
        self.time_limit = time_limit

    def _worker(self, func, queue):
        # 子进程中执行函数并返回结果、内存和时间数据
        start_time = time.perf_counter()
        process = psutil.Process()
        before_memory = process.memory_info().rss
        try:
            result = func()
        except Exception as e:
            queue.put(("error", e))
            return
        after_memory = process.memory_info().rss
        elapsed_time = time.perf_counter() - start_time
        queue.put(("success", result, elapsed_time, after_memory - before_memory))

    def run(self):
        queue = Queue()
        # 创建子进程运行目标函数
        process = multiprocessing.Process(target=self._worker, args=(self.func, queue))
        process.start()
        # 等待进程结束，超时则终止
        process.join(timeout=self.time_limit)

        if process.is_alive():
            # 超时：终止进程，记录日志，返回except_return
            process.terminate()
            process.join()  # 确保资源释放
            with open('Output.txt', mode='a', encoding='utf-8') as f:
                f.writelines("Time Limited Exceeded!\n")
            print("TimeoutError: Time Limited Exceeded!")
            return self.except_return
        else:
            # 正常结束：获取结果并处理
            data = queue.get()
            if data[0] == "error":
                print(data[1])
                return self.except_return
            result, elapsed_time, memory_used = data[1], data[2], data[3]
            with open('Output.txt', mode='a', encoding='utf-8') as f:
                f.writelines(f"Executing Time: {elapsed_time:.5f} \n")
                f.writelines(f"Memory Used: {memory_used} bytes \n")
            print(f"Executing Time: {elapsed_time:.5f}")
            print(f"Memory Used: {memory_used} bytes")
            return result

关键改动说明 #

• 用multiprocessing.Process代替线程：子进程独立运行目标函数，超时可调用terminate()强制终止
• 通过Queue传递数据：进程间无法直接共享变量，用队列传递执行结果、内存和时间数据
• 子进程内统计资源：在子进程中计算内存占用和执行时间，避免跨进程的资源统计误差
• 安全释放资源：超时终止进程后调用join()，确保子进程资源被正确回收

测试验证 #

用你提供的test()函数测试：

def test():
    time.sleep(3)
    print("3 seconds pass")
    time.sleep(3)
    print("6 seconds pass")
    return 6

if __name__ == '__main__':
    performance = Performance(test, 4, 0)
    a = performance.run()
    print(a)

输出与预期完全一致：

3 seconds pass
TimeoutError: Time Limited Exceeded!
0

内容的提问来源于stack exchange，提问作者Thế Hiểu Phạm