怎么在Python Tornado中正确使用async/await编写异步代码

在现代 Python Tornado 框架中正确使用 async/await 编写异步代码的核心原则是：将请求处理方法直接声明为 Python 原生的 async def，通过 await 挂起异步 I/O 释放事件循环，并彻底杜绝在协程中直接调用同步阻塞操作。现代 Tornado（6.0+）已经与 Python 标准库 asyncio 事件循环完全合体，不再需要使用旧版本的 @gen.coroutine 装饰器。只要遵循非阻塞规则，并为耗时的同步库配备线程池隔离，就能写出高吞吐、稳定且可维护的异步服务。

一、核心运行机制：Tornado 与现代 asyncio 的关系

Tornado 是 Python 领域历史悠久的异步 Web 框架，其核心优势在于单线程内依托事件循环（Event Loop）实现数以万计连接的并发处理。在 Python 3.5 引入原生的 async 和 await 关键字之前，Tornado 依赖生成器和 @gen.coroutine 配合 yield 来实现协程机制。

从 Tornado 5.0 开始，Tornado 的底层核心 IOLoop 已经直接重构为 Python 标准库 asyncio 事件循环的封装层；到了 Tornado 6.0 及以上版本，旧式基于 yield 的协程已全面退役，原生 async/await 成为标准唯一推荐的语法。

理解这一点对于写对异步代码至关重要：在 Tornado 中执行异步任务，本质上就是向当前线程的 asyncio 事件循环注册 Task 并挂起等待。当遇到 await 时，Tornado 暂停当前 Handler 的执行，将单线程的控制权交还给事件循环，从而继续处理其他连接的 HTTP 请求、数据包收发或定时器回调；直到被 await 的 Future 或 Task 完成，事件循环再唤醒当前协程继续往下执行。

二、正确编写 async/await 异步 Handler 的基础范式

在 Tornado 中定义一个异步路由处理器时，只需要遵循现代 Python 协程的标准写法。以下是正确编写的基础规则：

• 方法修饰直接声明为 async def：覆写 get()、post()、put() 等 HTTP 动作方法时，在函数名前加上 async 关键字。
• 严禁在 async def 上叠加 @gen.coroutine：这是从旧项目迁移或参考过时网文时最容易出现的错误。在 async def 方法上再添加 @gen.coroutine 会引发冗余的包装层，甚至导致返回值解析混乱。
• 在协程内部通过 await 调用异步对象：所有返回 Future 或本身是 async def 的协程调用，必须使用 await 关键字挂起。
• 响应输出逻辑：在方法内部正常调用 self.write(...) 写入响应体。在 async def 中，当协程执行完毕正常返回时，Tornado 会自动触发 self.finish()，一般不需要手动显式调用，除非涉及主动提前关闭或流式响应。

下面是一个标准的最小可用异步 Handler 示例：

import tornado.web
import tornado.ioloop
from tornado.httpclient import AsyncHTTPClient

class ModernAsyncHandler(tornado.web.RequestHandler):
    # 规则 1: 声明为 async def，切勿添加 @gen.coroutine 装饰器
    async def get(self):
        url = "https://httpbin.org/delay/1"
        
        # 规则 2: 使用 Tornado 原生的异步 HTTP 客户端
        client = AsyncHTTPClient()
        
        # 规则 3: 使用 await 挂起 I/O，等待期间事件循环去处理其他请求
        response = await client.fetch(url)
        
        # 规则 4: 正常写入响应，协程返回时框架自动处理 finish()
        self.write({
            "status": "ok",
            "bytes_received": len(response.body),
            "code": response.code
        })

三、改造演进步骤：从老旧写法迁移至原生 async/await

许多既有代码库中仍残留着基于 Tornado 4.x/5.x 的旧式写法。将代码平滑演进到现代纯净的 async/await 架构，可以按照以下规范化的五步流程进行改造和核对：

四、避免阻塞：根治“写了 async 却依然卡死”的陷阱

这是新手在 Tornado 中最常见的困惑：“我的 Handler 已经加了 async def，为什么发起一个耗时请求后，其他所有用户的请求都被卡住了？”

原因在于：async def 本身并不会自动将同步代码变成多线程或异步执行！ 如果协程内部执行了 CPU 密集型运算（如大批量图片压缩、复杂加密）或调用了同步阻塞的 I/O 函数（如标准库 time.sleep()、requests.get()、同步数据库客户端 pymysql / psycopg2），CPU 就会一直被该操作独占，事件循环根本没有机会去调用其他准备就绪的协程。

解决方案一：采用完全非阻塞的原生异步驱动

优先选择自身支持 await 的异步库。例如：

• 网络 HTTP 请求：使用 Tornado 内置的 AsyncHTTPClient，或者第三方的 httpx.AsyncClient、aiohttp。
• 延时休眠：使用 tornado.gen.sleep(seconds) 或 Python 标准库的 asyncio.sleep(seconds)，绝对不要用 time.sleep()。
• 数据库读写：使用专门的异步驱动，例如 MongoDB 使用 motor，PostgreSQL 使用 asyncpg，Redis 使用 redis.asyncio。

解决方案二：使用 ThreadPoolExecutor 隔离阻塞操作

在实际生产中，很多成熟的企业内部 SDK、加密算法库或传统 ORM 并没有提供异步版本。此时的正确策略是：通过 IOLoop.current().run_in_executor() 将同步代码推送到后台线程池运行，并通过 await 等待线程池返回结果。

五、高级用法：并发批处理、超时控制与离线后台任务

随着业务复杂度上升，一个 HTTP 请求往往需要聚合多个上游数据源，或者在请求返回后继续进行审计日志上报。以下是三种经典进阶模式：

1. 使用 asyncio.gather 并发发起多个异步子任务

如果需要同时查询三个独立的微服务或数据接口，切忌用三个连续的 await 串行执行。使用 asyncio.gather 可以并发执行这三个任务，总耗时将取决于最慢的一个接口，而不是三者之和：

import asyncio
from tornado.httpclient import AsyncHTTPClient

async def get(self):
    client = AsyncHTTPClient()
    urls = [
        "https://api.example.com/user_profile",
        "https://api.example.com/order_summary",
        "https://api.example.com/unread_messages"
    ]
    
    # 并发发出三个 HTTP 请求
    responses = await asyncio.gather(*(client.fetch(u) for u in urls))
    
    self.write({
        "profile": responses[0].body.decode("utf-8"),
        "orders": responses[1].body.decode("utf-8"),
        "messages": responses[2].body.decode("utf-8")
    })

2. 给异步操作添加严格的超时熔断

在微服务链条中，任何上游接口不可靠的响应都会耗尽本地等待资源。在 Tornado 中，可以直接使用 Python 标准库的 asyncio.wait_for 为任意协程设置超时：

import asyncio
import tornado.web

async def get(self):
    try:
        # 限制该异步操作必须在 2.5 秒内完成，否则抛出 TimeoutError
        res = await asyncio.wait_for(fetch_remote_data(), timeout=2.5)
        self.write({"result": res})
    except asyncio.TimeoutError:
        self.set_status(504)
        self.write({"error": "上游服务响应超时"})

3. 离开请求上下文的后台异步任务：spawn_callback

如果某些操作（例如发送通知邮件、推送统计埋点）不需要阻塞客户端得到结果，切勿在 Handler 内部 await。 应当使用 tornado.ioloop.IOLoop.current().spawn_callback(coro, *args) 将其作为独立任务丢给事件循环，随后 Handler 即可立刻响应用户并结束连接：

async def record_audit_log(user_id, action):
    # 模拟异步写入数据库或发送 MQ
    await tornado.gen.sleep(1)
    print(f"审计日志记录完毕: {user_id} - {action}")

class ActionHandler(tornado.web.RequestHandler):
    async def post(self):
        user_id = self.get_argument("user_id")
        
        # 立即返回响应给客户端
        self.write({"status": "success", "message": "请求已受理"})
        
        # 将耗时日志写入作为后台任务派发，不会延迟 HTTP 响应的交付
        tornado.ioloop.IOLoop.current().spawn_callback(
            record_audit_log, user_id, "ORDER_CREATED"
        )

六、完整可执行工程示例

为了让读者能够实际部署和验证，这里提供一个完整的、开箱即用的现代 Tornado 6 异步 Web 服务范例。代码涵盖了原生异步调用、线程池封装以及并发控制：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Tornado 现代 async/await 规范示例服务
依赖环境: Python >= 3.8, tornado >= 6.0
运行方式: python server.py
"""

import time
import asyncio
from concurrent.futures import ThreadPoolExecutor
import tornado.web
import tornado.ioloop
from tornado.httpclient import AsyncHTTPClient

# 全局共享线程池，用于执行无法异步化的第三方遗留库或 CPU 密集型任务
THREAD_POOL = ThreadPoolExecutor(max_workers=8)


def legacy_blocking_task(cost_seconds: float) -> str:
    """模拟无法改造为异步的传统同步阻塞库或密集计算"""
    time.sleep(cost_seconds)  # 此处在工作线程内阻塞，不会影响主事件循环
    return f"同步任务完成，用时 {cost_seconds} 秒"


class QuickPingHandler(tornado.web.RequestHandler):
    """快速探测接口，用于压测时验证主循环是否被卡住"""
    def get(self):
        self.write({"status": "healthy", "pong": True})


class NativeAsyncHandler(tornado.web.RequestHandler):
    """场景 A: 纯原生非阻塞异步 Handler"""
    async def get(self):
        client = AsyncHTTPClient()
        # await 期间事件循环自由轮询其他请求
        response = await client.fetch("https://httpbin.org/get")
        self.write({
            "mode": "native_async",
            "content_length": len(response.body),
            "http_status": response.code
        })


class BlockingAdaptedHandler(tornado.web.RequestHandler):
    """场景 B: 使用 run_in_executor 封装同步阻塞调用"""
    async def get(self):
        loop = tornado.ioloop.IOLoop.current()
        
        # 将阻塞函数委托至后台线程池，主循环在 await 处交出控制权
        result = await loop.run_in_executor(
            THREAD_POOL, 
            legacy_blocking_task, 
            1.5
        )
        self.write({
            "mode": "thread_pool_adapted",
            "message": result
        })


def make_app():
    return tornado.web.Application([
        (r"/ping", QuickPingHandler),
        (r"/api/async", NativeAsyncHandler),
        (r"/api/adapted", BlockingAdaptedHandler),
    ], debug=False)


if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    print("Tornado 服务已启动，监听端口: 8888")
    print("访问测试:")
    print("  curl http://127.0.0.1:8888/ping")
    print("  curl http://127.0.0.1:8888/api/async")
    print("  curl http://127.0.0.1:8888/api/adapted")
    
    # 启动 Tornado 主 IOLoop (底层即为 asyncio 事件循环)
    tornado.ioloop.IOLoop.current().start()

七、核心禁忌与选型决策指南

编写 Tornado 异步代码时，许多错误源于工具选型不当或对协程边界模糊。下面的决策矩阵与对照表列举了典型场景的最佳处理方案：

常见问题对照表

总结

在 Tornado 中正确编写异步代码的精髓在于“彻底的非阻塞”。掌握现代 Tornado（6.0+）的范式，可以概括为三句口诀：原生 async def 替代旧装饰器，I/O 操作必须显式 await，同步阻塞统统转交线程池。只要守住这一边界，Tornado 就能以极轻量的资源开销，提供极为出色的高并发性能。