241 lines
9.2 KiB
Markdown
241 lines
9.2 KiB
Markdown
# 同步/异步设计决策
|
|
|
|
本章详细说明在同步与异步编程模型上的设计决策，包括何时使用 `def`、何时使用 `async def`、`asyncio.to_thread` 的使用场景，以及需要避免的反模式。
|
|
|
|
:::tip 核心原则
|
|
FastAPI 基于 ASGI 异步框架，但并非所有端点都需要 `async def`。正确选择同步/异步模式，是避免事件循环阻塞、保证系统并发性能的关键。
|
|
:::
|
|
|
|
## FastAPI 的执行模型
|
|
|
|
FastAPI 对 `def` 和 `async def` 端点的处理方式不同：
|
|
|
|
```
|
|
async def endpoint():
|
|
# FastAPI 直接在事件循环中调用
|
|
# 适合 I/O 密集型操作（Redis、HTTP 调用、文件读取）
|
|
await some_async_operation()
|
|
|
|
def endpoint():
|
|
# FastAPI 自动在线程池（ThreadPoolExecutor）中调用
|
|
# 适合 CPU 密集型或同步 I/O 操作（数据库查询、文件处理）
|
|
some_sync_operation()
|
|
```
|
|
|
|
:::warning 重要认知
|
|
`def` 端点在 FastAPI 中**不会阻塞事件循环**，因为 FastAPI 会自动将其放入线程池执行。这意味着对于纯同步操作（如数据库查询），使用 `def` 是安全且推荐的。
|
|
:::
|
|
|
|
## 决策树
|
|
|
|
```
|
|
端点/Service 需要做什么？
|
|
│
|
|
├─ 包含 await 操作（Redis / 异步 HTTP / 异步文件读取）
|
|
│ └─ 使用 async def + await
|
|
│
|
|
├─ 需要 await parse_batch_ids(request)
|
|
│ └─ 使用 async def（批量删除端点）
|
|
│
|
|
├─ 纯同步操作（DB CRUD / 文件处理）
|
|
│ └─ 使用 def（FastAPI 自动线程池执行）
|
|
│
|
|
└─ 混合操作（同步 DB + 异步 Redis）
|
|
└─ 使用 async def + asyncio.to_thread() 包裹同步部分
|
|
```
|
|
|
|
## 三种场景详解
|
|
|
|
### 场景一：纯同步 — 使用 `def`
|
|
|
|
适用于纯数据库 CRUD 操作的简单模块（如 Position、Level、Link 等）。
|
|
|
|
```python
|
|
# src/api/v1/endpoints/position.py
|
|
@router.get('/page', summary='查询分页数据')
|
|
@permission_required("sys:position:page")
|
|
def page(request: Request):
|
|
"""同步端点：FastAPI 自动在线程池执行，不阻塞事件循环"""
|
|
return position_service.get_page(request)
|
|
|
|
@router.post('/add', summary='添加岗位')
|
|
@permission_required("sys:position:add")
|
|
@check_demo
|
|
def add(request: Request, data: PositionForm):
|
|
return position_service.add(request, data)
|
|
```
|
|
|
|
```python
|
|
# src/modules/system/position/service.py
|
|
class PositionService(BaseService[Position]):
|
|
# 所有方法均为同步 def
|
|
# 通过 Repository 访问数据库，Repository 内部使用同步 SQLAlchemy
|
|
...
|
|
```
|
|
|
|
:::tip 为什么同步操作不需要 async def？
|
|
FastAPI 对 `def` 端点会自动在线程池中执行，不阻塞事件循环。对于纯 DB CRUD 操作，使用 `def` 更简洁，且避免了不必要的协程开销。
|
|
:::
|
|
|
|
### 场景二：包含异步操作 — 使用 `async def`
|
|
|
|
适用于需要访问 Redis、异步 HTTP 调用、异步文件读取的场景。
|
|
|
|
```python
|
|
# src/modules/auth/service.py
|
|
async def login(request, data):
|
|
"""异步 Service：包含 Redis 操作（登录锁、Token 存储）"""
|
|
redis = get_redis()
|
|
|
|
# 检查登录失败锁定（异步 Redis 操作）
|
|
lock_key = f"login:lock:{data.username}"
|
|
is_locked = await redis.get(lock_key)
|
|
if is_locked:
|
|
return R.failed("账号已锁定，请稍后再试")
|
|
|
|
# 验证用户名密码（同步 DB 操作，用 to_thread 包裹）
|
|
user = await asyncio.to_thread(
|
|
user_repo.get_one, username=data.username
|
|
)
|
|
if not user or not verify_password(data.password, user.password):
|
|
# 记录失败次数（异步 Redis 操作）
|
|
count = await incr_with_expire(redis, f"login:fail:{data.username}", 300)
|
|
if count >= 5:
|
|
await redis.set(lock_key, 1, ex=600)
|
|
return R.failed("用户名或密码错误")
|
|
|
|
# 生成 JWT Token（同步操作）
|
|
token = create_access_token(user.id, user.username)
|
|
|
|
# 存储 Token 到 Redis（异步操作）
|
|
await redis.set(f"token:{user.id}", token, ex=JWT_EXPIRE_MINUTES * 60)
|
|
|
|
return R.ok(data={"token": token, ...})
|
|
```
|
|
|
|
### 场景三：混合操作 — 使用 `async def` + `asyncio.to_thread`
|
|
|
|
当 `async def` 函数中需要执行同步阻塞操作时，使用 `asyncio.to_thread()` 将其放入线程池。
|
|
|
|
```python
|
|
# src/modules/auth/service.py
|
|
async def login(request, data):
|
|
# 同步 DB 查询 → 用 to_thread 包裹
|
|
user = await asyncio.to_thread(
|
|
user_repo.get_one, username=data.username
|
|
)
|
|
|
|
# 同步密码验证 → 用 to_thread 包裹
|
|
is_valid = await asyncio.to_thread(
|
|
verify_password, data.password, user.password
|
|
)
|
|
|
|
# 异步 Redis 操作 → 直接 await
|
|
redis = get_redis()
|
|
await redis.set(f"token:{user.id}", token, ex=3600)
|
|
```
|
|
|
|
:::warning to_thread 使用要点
|
|
`asyncio.to_thread(func, *args)` 会将同步函数放入默认线程池执行。注意：
|
|
1. 传递函数引用和参数，不要传递函数调用结果
|
|
2. 正确：`await asyncio.to_thread(repo.get_one, username='admin')`
|
|
3. 错误：`await asyncio.to_thread(repo.get_one(username='admin'))`
|
|
:::
|
|
|
|
## 各模块的同步/异步选择
|
|
|
|
| 模块 | Service 类型 | 原因 |
|
|
|:---|:---|:---|
|
|
| Position / Level / Link | 同步 `def` | 纯 DB CRUD |
|
|
| Dept / Menu / Category | 同步 `def` | 纯 DB CRUD（含树形查询） |
|
|
| Role / User | 同步 `def` | 纯 DB CRUD（含关联查询） |
|
|
| Article / Notice | 同步 `def` | DB CRUD + 文件处理 |
|
|
| Auth (login/logout) | 异步 `async def` | Redis 操作（锁定/Token/黑名单） |
|
|
| Dict / Config | 同步 `def` | DB CRUD + 进程内缓存 |
|
|
| Job (scheduler) | 混合 | DB 查询 + Redis 分布式锁 |
|
|
| 批量删除端点 | 异步 `async def` | `await parse_batch_ids(request)` |
|
|
|
|
## 批量删除端点的特殊性
|
|
|
|
:::tip 为什么批量删除端点必须是 async def？
|
|
批量删除端点需要 `await parse_batch_ids(request)` 异步读取请求体。这是 FastAPI + Starlette 的限制：`request.body()` 是异步方法，必须在 `async def` 中调用。
|
|
:::
|
|
|
|
```python
|
|
# src/api/v1/endpoints/position.py
|
|
@router.delete('/batchDelete', summary='批量删除岗位')
|
|
@permission_required("sys:position:batchDelete")
|
|
@check_demo
|
|
async def batch_delete(request: Request):
|
|
return await position_service.batch_delete(request)
|
|
```
|
|
|
|
```python
|
|
# src/core/base_service.py
|
|
class BaseService:
|
|
async def batch_delete(self, request) -> R:
|
|
"""批量删除：请求体为 ID 数组 [1,2,3]，线程池执行避免阻塞事件循环"""
|
|
ids, err = await parse_batch_ids(request) # 异步读取请求体
|
|
if err:
|
|
return err
|
|
return await asyncio.to_thread(self.delete, ids) # 同步删除放线程池
|
|
```
|
|
|
|
注意：Service 层的 `batch_delete` 是 `async def`，但它内部通过 `asyncio.to_thread` 将实际的删除操作（`self.delete`）放入线程池执行，避免阻塞事件循环。`self.delete` 内部会调用 `self._before_delete(ids)` 钩子做前置校验，再调用 `batch_delete_with_r(self.repo, ids)` 完成软删除。
|
|
|
|
## 禁止的模式
|
|
|
|
:::danger 禁止在 async def 中直接调用同步 DB 操作
|
|
在 `async def` 函数中直接调用同步 DB 查询会阻塞事件循环，导致整个应用在该请求完成前无法处理其他请求。
|
|
:::
|
|
|
|
### 错误示例
|
|
|
|
```python
|
|
# 错误！在 async def 中直接调用同步 DB 操作
|
|
async def get_user(request, user_id):
|
|
user = user_repo.get_by_id(user_id) # 阻塞事件循环！
|
|
return R.ok(data=user.to_dict())
|
|
```
|
|
|
|
### 正确示例
|
|
|
|
```python
|
|
# 正确：用 asyncio.to_thread 包裹
|
|
async def get_user(request, user_id):
|
|
user = await asyncio.to_thread(user_repo.get_by_id, user_id)
|
|
return R.ok(data=user.to_dict())
|
|
```
|
|
|
|
```python
|
|
# 正确：改用同步 def（推荐，更简洁）
|
|
def get_user(request, user_id):
|
|
user = user_repo.get_by_id(user_id) # FastAPI 自动线程池执行
|
|
return R.ok(data=user.to_dict())
|
|
```
|
|
|
|
## asyncio.to_thread 使用场景
|
|
|
|
| 场景 | 处理方式 |
|
|
|:---|:---|
|
|
| `async def` 中调用同步 DB 查询 | `await asyncio.to_thread(repo.get_by_id, id)` |
|
|
| `async def` 中调用同步密码验证 | `await asyncio.to_thread(verify_password, pwd, hash)` |
|
|
| `async def` 中调用同步文件操作 | `await asyncio.to_thread(save_file, content, path)` |
|
|
| `async def` 中调用同步加密操作 | `await asyncio.to_thread(hash_password, pwd)` |
|
|
| `def` 中的任何同步操作 | 直接调用，FastAPI 自动线程池执行 |
|
|
|
|
## 同步/异步选择速查表
|
|
|
|
| 条件 | 选择 | 原因 |
|
|
|:---|:---|:---|
|
|
| 只有 DB CRUD | `def` | FastAPI 自动线程池，简洁安全 |
|
|
| 有 Redis 操作 | `async def` | Redis 异步客户端需要 `await` |
|
|
| 有 `parse_batch_ids` | `async def` | 请求体读取需要 `await` |
|
|
| 有异步 HTTP 调用 | `async def` | 外部 API 调用需要 `await` |
|
|
| 混合同步+异步 | `async def` + `to_thread` | 同步部分用线程池包裹 |
|
|
| 定时任务回调 | 同步 `def` | APScheduler 在独立线程执行 |
|
|
|
|
## 总结
|
|
|
|
同步/异步的选择遵循一个简单原则：**有 `await` 操作用 `async def`，否则用 `def`**。FastAPI 对 `def` 端点自动使用线程池执行，不会阻塞事件循环，因此纯 DB CRUD 模块无需强制使用 `async def`。当 `async def` 中需要调用同步阻塞操作时，使用 `asyncio.to_thread()` 将其放入线程池。禁止在 `async def` 中直接调用同步 DB 操作，这是导致事件循环阻塞、并发性能下降的常见错误。
|