FastAPI中实现work函数串行调用，统计async.sleep()执行耗时

要让work函数同一时间仅能被调用一次，同时保证多次调用test接口时，各响应的time字段值与前一次至少相差3秒，我们可以用**异步锁（asyncio.Lock）**来实现串行化执行，具体代码修改如下：

import asyncio
from time import monotonic
# 假设router和TestResponse已提前定义
router = ...
TestResponse = ...

# 定义全局异步锁，控制work函数的串行执行
work_lock = asyncio.Lock()

async def work():
    # 修正原代码错误：异步休眠必须加await
    await asyncio.sleep(3)

@router.get('')
async def test():
    time1 = monotonic()
    # 自动获取/释放锁，同一时间仅一个协程能执行work
    async with work_lock:
        await work()
    time2 = monotonic()
    return TestResponse(time=time2-time1)

关键说明： #

• 全局work_lock是协程安全的异步锁，所有请求会排队等待锁释放后再执行work，从根源避免work被并发调用。
• 原代码的asyncio.sleep(3)缺少await，这是异步代码的常见疏漏，必须添加才能让协程真正休眠3秒。
• async with语法会自动处理锁的获取与释放，避免手动操作时出现锁泄漏问题。

当多个请求同时调用test接口时，第一个请求会立即执行work（耗时3秒），后续请求会依次排队等待锁释放，每个请求的time字段值都会比前一个至少大3秒，完全满足需求。

内容的提问来源于stack exchange，提问作者Hahan't