Python中asyncio.Condition未生效，关键API仍重复调用排查

以下是导致你遇到的「所有账号都触发API调用」问题的常见原因，以及对应的修正思路：

1. 未在锁保护下操作缓存 #

asyncio.Condition的核心是锁+条件通知，所有读写缓存字典的操作必须在async with cond:上下文内执行。如果修改缓存时没持有锁，其他协程可能看不到最新的缓存数据，导致条件判断失效。

比如错误写法：

# 错误：修改缓存时没加锁
cache[account_id] = data
cond.notify_all()

正确写法必须包裹在锁内：

async with cond:
    cache[account_id] = data
    cond.notify_all()

2. 等待条件前未先检查缓存状态 #

很多人会直接调用await cond.wait()，但正确逻辑应该是先检查缓存，不存在再等待。否则当缓存已经被其他协程填充后，当前协程还会无意义地等待通知，甚至错过缓存数据直接走API调用逻辑。

错误写法：

async with cond:
    await cond.wait()
    # 才检查缓存，此时可能已经错过更新
    if account_id not in cache:
        data = await call_api()
        cache[account_id] = data

正确写法（循环检查+等待，避免虚假唤醒）：

async with cond:
    while account_id not in cache:
        await cond.wait()
    return cache[account_id]

3. 缓存更新后未触发通知 #

当PAYER账号获取到子账号数据并更新缓存后，必须调用cond.notify_all()（或cond.notify()）唤醒所有等待的协程。如果漏掉这一步，其他子账号任务会一直阻塞在wait状态，直到超时（如果设置了），甚至直接跳过等待去调用API。

4. 任务逻辑未阻塞等待缓存，直接并行执行 #

如果你的代码里，协程只是简单检查缓存，不存在就直接调用API，没有用Condition的wait来阻塞，那么所有任务会同时启动，此时缓存为空，所有任务都会触发API调用。

修正后的核心逻辑示例 #

import asyncio

# 全局缓存字典
account_cache = {}
# 条件变量
cache_cond = asyncio.Condition()

async def get_account_data(account_id):
    async with cache_cond:
        # 先检查缓存，存在直接返回
        if account_id in account_cache:
            return account_cache[account_id]
        
        # 处理PAYER账号，获取所有子账号数据
        if account_id == "PAYER":
            # 模拟调用API获取 payer 及其子账号数据
            payer_info = await fetch_payer_api()
            # 更新缓存：存入payer本身
            account_cache[account_id] = payer_info["self_data"]
            # 存入所有子账号数据
            for sub_acc, sub_data in payer_info["sub_accounts"].items():
                account_cache[sub_acc] = sub_data
            # 唤醒所有等待的子账号任务
            cache_cond.notify_all()
            return payer_info["self_data"]
        else:
            # 子账号：循环等待缓存被填充
            while account_id not in account_cache:
                await cache_cond.wait()
            return account_cache[account_id]

async def fetch_payer_api():
    # 模拟API调用延迟
    await asyncio.sleep(1)
    return {
        "self_data": {"type": "payer", "balance": 1000},
        "sub_accounts": {
            "SUB001": {"type": "sub", "balance": 200},
            "SUB002": {"type": "sub", "balance": 300}
        }
    }

async def main():
    target_accounts = ["PAYER", "SUB001", "SUB002"]
    tasks = [get_account_data(acc) for acc in target_accounts]
    results = await asyncio.gather(*tasks)
    for acc_id, data in zip(target_accounts, results):
        print(f"账号 {acc_id} 数据: {data}")

asyncio.run(main())

这个示例中：

• 所有缓存操作都在锁保护下执行，保证协程安全
• 子账号任务会循环等待缓存，直到被PAYER任务唤醒
• PAYER任务更新缓存后立即触发通知，避免其他任务重复调用API
• 最终只有PAYER会调用API，子账号直接复用缓存数据

内容的提问来源于stack exchange，提问作者Bumblebee