签入版本
This commit is contained in:
@@ -0,0 +1,194 @@
|
||||
# Tornado IOLoop 与异步模型
|
||||
|
||||
本章详细说明 Tornado 的异步编程模型，包括 IOLoop 事件循环、协程使用方式、与同步 SQLAlchemy 的协作，以及需要避免的反模式。
|
||||
|
||||
:::tip 核心原则
|
||||
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` 中直接调用：
|
||||
|
||||
```python
|
||||
# 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)
|
||||
```
|
||||
|
||||
```python
|
||||
# src/modules/link/service.py
|
||||
class LinkService(BaseService):
|
||||
# 同步方法，直接调用 Repository
|
||||
def get_page(self, handler):
|
||||
# Repository 内部使用同步 SQLAlchemy Session
|
||||
return self.repo.paginate(...)
|
||||
```
|
||||
|
||||
:::warning 为什么同步 DB 操作在 Tornado 中可行？
|
||||
本项目的同步 SQLAlchemy 操作通常执行很快（毫秒级），短暂阻塞 IOLoop 在实际业务中可接受。如果遇到慢查询场景，可考虑使用 `tornado.ioloop.IOLoop.current().run_in_executor()` 将其放入线程池。
|
||||
:::
|
||||
|
||||
## Handler 方法的异步模式
|
||||
|
||||
### 标准模式：async def
|
||||
|
||||
所有 Handler 方法默认使用 `async def`：
|
||||
|
||||
```python
|
||||
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 操作一样是同步调用：
|
||||
|
||||
```python
|
||||
# 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 启动与关闭
|
||||
|
||||
### 启动流程
|
||||
|
||||
```python
|
||||
# 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()
|
||||
```
|
||||
|
||||
### 优雅关闭
|
||||
|
||||
```python
|
||||
def _shutdown():
|
||||
"""收到退出信号时关闭调度器并停止事件循环。"""
|
||||
from modules.job.scheduler import shutdown_scheduler
|
||||
shutdown_scheduler()
|
||||
ioloop.stop()
|
||||
```
|
||||
|
||||
## 禁止的模式
|
||||
|
||||
:::danger 避免在 Handler 中执行长时间同步操作
|
||||
在 `async def` 中执行长时间同步操作（如大文件读写、复杂计算）会阻塞整个 IOLoop，导致所有并发请求被阻塞。
|
||||
:::
|
||||
|
||||
### 错误示例
|
||||
|
||||
```python
|
||||
# 错误！在 async def 中执行长时间同步操作
|
||||
async def get(self):
|
||||
# 大文件读取阻塞 IOLoop
|
||||
with open("huge_file.csv", "r") as f:
|
||||
data = f.read() # 阻塞！
|
||||
self.finish({"code": 0, "data": data})
|
||||
```
|
||||
|
||||
### 正确示例
|
||||
|
||||
```python
|
||||
# 正确：使用 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 影响。
|
||||
Reference in New Issue
Block a user