Python异步编程中如何快速获取大量请求中的第一个有效返回值？

import asyncio
import aiohttp
from typing import Any, Optional

async def fetch_first_valid(
    urls: list[str],
    session: aiohttp.ClientSession,
    timeout: float = 5.0
) -> Optional[Any]:
    """
    并发请求多个URL，返回第一个成功的响应内容
    """
    async def fetch_one(url: str) -> str:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=timeout)) as resp:
            resp.raise_for_status()
            return await resp.text()
    
    tasks = [asyncio.create_task(fetch_one(url)) for url in urls]
    
    # 使用asyncio.wait等待第一个完成的任务
    done, pending = await asyncio.wait(
        tasks,
        return_when=asyncio.FIRST_COMPLETED
    )
    
    # 取消剩余未完成的任务
    for task in pending:
        task.cancel()
    
    # 获取第一个完成的结果
    if done:
        first_task = next(iter(done))
        try:
            return first_task.result()
        except Exception:
            # 如果第一个任务失败，可以继续等待其他任务
            # 这里简单返回None，实际可根据需求调整
            pass
    
    # 等待所有任务结束（避免cancel警告）
    await asyncio.gather(*pending, return_exceptions=True)
    return None

async def main():
    urls = [
        "https://api1.example.com/data",
        "https://api2.example.com/endpoint",
        "https://backup.example.com/info"
    ]
    
    async with aiohttp.ClientSession() as session:
        result = await fetch_first_valid(urls, session)
        print(f"First valid response: {result}")

if __name__ == "__main__":
    asyncio.run(main())

核心思路就是用asyncio.wait()的FIRST_COMPLETED参数，配合任务取消机制。这里有几个关键点：

1. 并发创建所有请求任务：用列表推导式同时创建所有请求的Task对象
2. 等待第一个完成：asyncio.wait()设置return_when=asyncio.FIRST_COMPLETED，一旦有任务完成就返回
3. 清理剩余任务：立即取消其他还在进行的请求，避免资源浪费
4. 异常处理：第一个完成的请求可能失败，这时可以继续等待其他任务或直接返回

如果第一个请求失败想继续等下一个，可以改成循环处理。不过通常多个备用API的场景，第一个能成的概率就很高了。

用FIRST_COMPLETED比挨个等快多了，特别是当备用接口响应时间差异大时。