如何为Tornado中@run_on_executor装饰的函数设置超时条件？

@run_on_executor本身没有内置超时配置参数，最优方案就是配合tornado.gen.with_timeout直接使用。被@run_on_executor装饰的方法返回的是标准可等待Future对象，完全兼容with_timeout的入参要求，不需要额外做适配改造。

改造后的代码示例 #

import json
import time
import tornado.web
from concurrent.futures import ThreadPoolExecutor
from tornado.concurrent import run_on_executor
from tornado.gen import with_timeout
from tornado.util import TimeoutError

class MyHandler(tornado.web.RequestHandler):

    def initialize(self) -> None:
        self.executor = ThreadPoolExecutor(1)

    @run_on_executor
    def blocking_function(self) -> None:
        """ Run Blocking Function on ThreadPoolExecutor. """
        seconds = 10
        time.sleep(seconds)
        response = json.dumps({"message": f"Slept for {seconds} seconds."})
        return response

    async def get(self) -> None:
        try:
            # 此处设置超时时间为3秒，可根据业务需求调整
            response = await with_timeout(
                timeout=3,
                future=self.blocking_function()
            )
            self.write(response)
        except TimeoutError:
            # 自定义超时返回逻辑
            self.set_status(504)
            self.write(json.dumps({"message": "Request processing timed out"}))

核心注意事项 #

• 超时仅终止Tornado侧的等待逻辑，不会强制中止线程池中已经启动的任务：Python标准库的ThreadPoolExecutor没有原生终止运行中线程的能力，如果需要终止任务执行，需要在阻塞函数内部自行实现终止标记判断逻辑，比如定期检查实例级别的停止标记位。
• 若有全局统一的超时要求，可以把with_timeout逻辑封装成独立装饰器，包裹被@run_on_executor装饰的方法，避免每次await时重复编写超时逻辑。
• 当使用单线程ThreadPoolExecutor时，超时后未完成的任务会占用唯一的工作线程，后续提交的任务会进入排队队列，需要根据业务场景合理设置线程池大小，避免大量超时任务占满线程池导致服务不可用。

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