为计算器代码设置超时：解决大数值计算冻结及任务无法取消问题

问题根源 #

你当前代码失效的核心原因是：do_calculation里的sympy计算是同步阻塞操作，哪怕函数标了async，但里面没有任何await语句，本质还是同步代码。这种情况下，事件循环会被同步计算完全占住，根本没时间处理超时定时器，直到计算结束才会响应——自然达不到3秒超时的效果。就算触发了超时，task.cancel()也没用，因为同步任务没有await点来响应取消信号。

解决方案 #

要让超时机制生效，必须把同步计算从事件循环线程里移出去，用线程池或进程池来执行，让事件循环能自由处理超时逻辑。

方案1：用线程池（Python 3.9+） #

把同步计算改成普通函数，用asyncio.to_thread丢到线程池执行：

import asyncio
import sympy
# 假设numerize是你已有的函数

def do_calculation(realcalculation, X, Y, A, B, C):
    try:
        result = str(numerize(sympy.sympify(realcalculation).subs(dict(X=X, Y=Y, A=A, B=B, C=C))))
    except Exception:
        return "NAN"
    return result

async def calculate(realcalculation, X, Y, A, B, C):
    task = asyncio.create_task(asyncio.to_thread(do_calculation, realcalculation, X, Y, A, B, C))
    try:
        result = await asyncio.wait_for(task, timeout=3)
    except asyncio.TimeoutError:
        task.cancel()
        result = "NAN"
    return result

# 调用方式不变
result = await calculate(realcalculation, X, Y, A, B, C)

方案2：用线程池（兼容Python 3.8及以下） #

如果你的Python版本低于3.9，用loop.run_in_executor：

import asyncio
import sympy

def do_calculation(realcalculation, X, Y, A, B, C):
    try:
        result = str(numerize(sympy.sympify(realcalculation).subs(dict(X=X, Y=Y, A=A, B=B, C=C))))
    except Exception:
        return "NAN"
    return result

async def calculate(realcalculation, X, Y, A, B, C):
    loop = asyncio.get_running_loop()
    future = loop.run_in_executor(None, do_calculation, realcalculation, X, Y, A, B, C)
    try:
        result = await asyncio.wait_for(future, timeout=3)
    except asyncio.TimeoutError:
        future.cancel()
        result = "NAN"
    return result

方案3：用进程池（彻底终止计算） #

线程池的局限性是：无法强制终止后台线程，超时后计算可能还在后台跑，浪费资源。如果要彻底终止计算，用进程池：

import asyncio
import sympy
import multiprocessing

def do_calculation(realcalculation, X, Y, A, B, C):
    try:
        result = str(numerize(sympy.sympify(realcalculation).subs(dict(X=X, Y=Y, A=A, B=B, C=C))))
    except Exception:
        return "NAN"
    return result

async def calculate(realcalculation, X, Y, A, B, C):
    loop = asyncio.get_running_loop()
    # 注意：进程池建议全局初始化，避免重复创建销毁开销
    with multiprocessing.Pool(1) as pool:
        future = loop.run_in_executor(pool, do_calculation, realcalculation, X, Y, A, B, C)
        try:
            result = await asyncio.wait_for(future, timeout=3)
        except asyncio.TimeoutError:
            pool.terminate()  # 强制终止进程，彻底停止计算
            result = "NAN"
    return result

关键说明 #

• 把同步计算移到线程/进程池后，事件循环不再被阻塞，3秒超时定时器能正常触发。
• 线程池适合轻量到中等计算量，进程池适合超大计算量（比如9⁹⁹⁹⁹⁹⁹这种），能彻底终止计算进程，避免资源浪费。
• 原代码里do_calculation的try-except写法有问题：捕获所有异常后又raise，导致返回"NAN"后还会抛出异常，修改后直接返回"NAN"即可，无需再抛出。

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