Tornado IOLoop 与异步模型

本章详细说明 Tornado 的异步编程模型，包括 IOLoop 事件循环、协程使用方式、与同步 SQLAlchemy 的协作，以及需要避免的反模式。

核心原则

Tornado 基于单线程事件循环（IOLoop），Handler 方法默认为协程（async def）。正确理解 IOLoop 与同步阻塞操作的关系，是避免事件循环阻塞、保证系统并发性能的关键。

Tornado 的执行模型

Tornado 使用单线程 IOLoop 事件循环处理请求：

IOLoop 事件循环
   │
   ├─ 请求1 进入 → Handler.get() 协程 → await → 挂起
   │                                              │
   ├─ 请求2 进入 → Handler.get() 协程 → await → 挂起
   │                                              │
   ├─ 请求1 的 await 完成 → 恢复执行 → 返回响应
   │
   ├─ 请求2 的 await 完成 → 恢复执行 → 返回响应
   │
   └─ ...

Tornado 不会自动将同步函数放入线程池。在 async def 中直接调用同步阻塞操作会阻塞整个 IOLoop。本项目的 DB 和 Redis 操作均为同步调用，通常毫秒级完成，短暂阻塞在实际业务中可接受。

决策树

Handler/Service 需要做什么？
│
├─ 包含 await 操作（Redis / 异步 HTTP / 异步文件读取）
│   └─ 使用 async def + await
│
├─ 纯同步操作（DB CRUD / 文件处理）
│   └─ 使用 async def（Tornado Handler 默认）+ 直接调用同步代码
│
└─ 混合操作（同步 DB + 异步 Redis）
    └─ 使用 async def，同步部分直接调用，异步部分 await

Tornado 与 SQLAlchemy 的协作

本项目使用 同步 SQLAlchemy（scoped_session），在 Tornado 的 async def 中直接调用：

# src/modules/link/handlers.py
class LinkPageHandler(BaseHandler):
    @permission_required("sys:link:list")
    async def get(self):
        # 直接调用同步 Service（内部使用同步 SQLAlchemy）
        return await link_service.get_page(self)

# src/modules/link/service.py
class LinkService(BaseService):
    # 同步方法，直接调用 Repository
    def get_page(self, handler):
        # Repository 内部使用同步 SQLAlchemy Session
        return self.repo.paginate(...)

为什么同步 DB 操作在 Tornado 中可行？

本项目的同步 SQLAlchemy 操作通常执行很快（毫秒级），短暂阻塞 IOLoop 在实际业务中可接受。如果遇到慢查询场景，可考虑使用 tornado.ioloop.IOLoop.current().run_in_executor() 将其放入线程池。

Handler 方法的异步模式

标准模式：async def

所有 Handler 方法默认使用 async def：

class LinkAddHandler(BaseHandler):
    @permission_required("sys:link:add")
    @check_demo
    @operation_log("友链管理", "添加")
    async def post(self):
        return await link_service.add(self)

涉及 Redis 的同步操作

本项目使用 同步 Redis 客户端（from core.redis import redis），Redis 操作与 DB 操作一样是同步调用：

# src/modules/auth/service.py
from core.redis import redis

async def login(self):
    # 检查登录失败锁定（同步 Redis 操作）
    lock_key = f"login:lock:{ip}:{username}"
    is_locked = redis.get(lock_key)
    if is_locked:
        return R.failed(self, "账号已锁定，请稍后再试")

    # 验证用户名密码（同步 DB 操作）
    user = UserRepository().get_by_username(username)
    if not user:
        # 记录失败次数（同步 Redis 操作）
        redis.incr(f"login:fail:{ip}:{username}")
        redis.expire(f"login:fail:{ip}:{username}", 300)
        return R.failed(self, "用户名或密码错误")

    # 签发 JWT Token（同步操作）
    token = create_token({"userId": user.id})

    return R.ok(self, msg="登录成功", data={"access_token": token})

各模块的异步使用模式

模块	模式	原因
Link / Position / Level	async def + 同步 DB	纯 DB CRUD，同步操作快速返回
Dept / Menu / Category	async def + 同步 DB	纯 DB CRUD（含树形查询）
Role / User	async def + 同步 DB	纯 DB CRUD（含关联查询）
Article / Notice	async def + 同步 DB + 文件	DB CRUD + 文件处理
Auth (login/logout)	async def + 混合	同步 DB + 异步 Redis
Dict / Config	async def + 同步 DB	DB CRUD + Redis 缓存
Job (scheduler)	同步 def	APScheduler 在独立线程执行

IOLoop 启动与关闭

启动流程

# src/bootstrap.py
def main():
    app = create_app()
    http_server = tornado.httpserver.HTTPServer(app)
    http_server.listen(options.port)

    # 启动定时任务调度器
    from modules.job.scheduler import init_scheduler
    init_scheduler()

    ioloop = tornado.ioloop.IOLoop.current()

    # 注册信号处理（优雅关闭）
    for sig in (signal.SIGINT, signal.SIGTERM):
        signal.signal(sig, lambda s, f: ioloop.add_callback_from_signal(_shutdown))

    ioloop.start()

优雅关闭

def _shutdown():
    """收到退出信号时关闭调度器并停止事件循环。"""
    from modules.job.scheduler import shutdown_scheduler
    shutdown_scheduler()
    ioloop.stop()

禁止的模式

避免在 Handler 中执行长时间同步操作

在 async def 中执行长时间同步操作（如大文件读写、复杂计算）会阻塞整个 IOLoop，导致所有并发请求被阻塞。

错误示例

# 错误！在 async def 中执行长时间同步操作
async def get(self):
    # 大文件读取阻塞 IOLoop
    with open("huge_file.csv", "r") as f:
        data = f.read()  # 阻塞！
    self.finish({"code": 0, "data": data})

正确示例

# 正确：使用 run_in_executor 将阻塞操作放入线程池
async def get(self):
    import asyncio
    loop = asyncio.get_event_loop()
    data = await loop.run_in_executor(None, self._read_file, "huge_file.csv")
    self.finish({"code": 0, "data": data})

def _read_file(self, path):
    with open(path, "r") as f:
        return f.read()

总结

Tornado 基于单线程 IOLoop 事件循环，Handler 方法默认 async def。本项目使用同步 SQLAlchemy，DB 操作通常毫秒级，直接在协程中调用可接受。涉及 Redis 等异步操作时使用 await。避免在协程中执行长时间同步操作，必要时使用 run_in_executor 放入线程池。定时任务调度器在独立线程运行，不受 IOLoop 影响。