同步/异步设计决策

本章详细说明在同步与异步编程模型上的设计决策，包括何时使用 def、何时使用 async def、asyncio.to_thread 的使用场景，以及需要避免的反模式。

核心原则

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()

重要认知

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 等）。

# 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)

# src/modules/system/position/service.py
class PositionService(BaseService[Position]):
 # 所有方法均为同步 def
 # 通过 Repository 访问数据库，Repository 内部使用同步 SQLAlchemy
 ...

为什么同步操作不需要 async def？

FastAPI 对 def 端点会自动在线程池中执行，不阻塞事件循环。对于纯 DB CRUD 操作，使用 def 更简洁，且避免了不必要的协程开销。

场景二：包含异步操作 — 使用 async def

适用于需要访问 Redis、异步 HTTP 调用、异步文件读取的场景。

# 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() 将其放入线程池。

# 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)

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)

批量删除端点的特殊性

为什么批量删除端点必须是 async def？

批量删除端点需要 await parse_batch_ids(request) 异步读取请求体。这是 FastAPI + Starlette 的限制：request.body() 是异步方法，必须在 async def 中调用。

# 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)

# 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) 完成软删除。

禁止的模式

禁止在 async def 中直接调用同步 DB 操作

在 async def 函数中直接调用同步 DB 查询会阻塞事件循环，导致整个应用在该请求完成前无法处理其他请求。

错误示例

# 错误！在 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())

正确示例

# 正确：用 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())

# 正确：改用同步 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 操作，这是导致事件循环阻塞、并发性能下降的常见错误。