Tornado 6.1 实现POST响应返回后后台执行长耗时任务不阻塞服务

你遇到的阻塞问题本质是Tornado的IO事件循环是单线程运行的，直接在请求处理逻辑里调用同步阻塞函数，哪怕是在self.finish()之后，依然会卡住主线程，导致其他请求无法被及时处理。你可以根据你的阻塞逻辑类型，选以下两种方案实现：

方案1：阻塞逻辑为同步代码（比如用time.sleep、requests调用第三方接口） #

用IOLoop.run_in_executor把阻塞任务提交到独立线程池运行，不会占用主线程资源。

修改后的代码示例 #

import tornado.web
import time
import logging
from concurrent.futures import ThreadPoolExecutor
from tornado.ioloop import IOLoop

# 你原来的validateInFolder、post等函数保持不变

class AddHandler(tornado.web.RequestHandler):
    # 初始化线程池，max_workers根据你的并发需求调整，IO密集型任务建议设为CPU核心数的4-8倍
    executor = ThreadPoolExecutor(max_workers=8)

    def initialize(self, folderpaths):
        self.folderpaths = folderpaths

    def blockingFunction(self):
        time.sleep(320)
        post("AWAKE")

    def post(self):
        user = self.get_argument('user')
        folderpath = self.get_argument('inpath')
        outpath = self.get_argument('outpath')
        workflow_value = self.get_argument('workflow')
        status_code, status_text = validateInFolder(folderpath)
        if (status_code == 200):
            logging.info("Status Code 200")
            result = self.folderpaths.add_file(user, folderpath, outpath, workflow_value)
            self.write(result)
            self.finish()
            # 提交阻塞任务到线程池后台运行，不阻塞主线程
            IOLoop.current().run_in_executor(self.executor, self.blockingFunction)

注意事项 #

• 如果你的阻塞逻辑是CPU密集型，把ThreadPoolExecutor换成ProcessPoolExecutor即可，避免Python GIL导致的性能问题。
• 如果阻塞函数中需要操作Tornado主线程的资源（比如修改共享变量、调用请求上下文相关方法），需要用IOLoop.current().add_callback()把对应逻辑丢回主线程执行，避免线程安全问题。

方案2：阻塞逻辑为异步代码（比如用aiohttp调用第三方接口、asyncio.sleep） #

如果已经把blockingFunction改成了async异步方法，直接调用不await会触发警告，你可以用asyncio.create_task把它包装成后台异步任务，不需要await也不会卡主线程。

代码示例 #

import asyncio
import tornado.web
import logging
import aiohttp

class AddHandler(tornado.web.RequestHandler):
    def initialize(self, folderpaths):
        self.folderpaths = folderpaths

    async def blockingFunction(self):
        await asyncio.sleep(320)
        async with aiohttp.ClientSession() as session:
            await session.post("你的接口地址", data="AWAKE")

    async def post(self):
        user = self.get_argument('user')
        folderpath = self.get_argument('inpath')
        outpath = self.get_argument('outpath')
        workflow_value = self.get_argument('workflow')
        status_code, status_text = validateInFolder(folderpath)
        if (status_code == 200):
            logging.info("Status Code 200")
            result = self.folderpaths.add_file(user, folderpath, outpath, workflow_value)
            self.write(result)
            self.finish()
            # 提交异步任务到事件循环后台运行，不需要await，无警告
            asyncio.create_task(self.blockingFunction())

生产环境补充建议 #

如果你的后台任务要求高可靠，不能因为服务重启、崩溃丢失，建议不要用进程/线程池临时运行任务，改用独立的任务队列组件，请求验证通过后把任务参数丢到队列里，由独立的worker进程消费执行，稳定性更高。

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