python中并发代码执行

1.asyncio.gather() —— 统一收集结果

import asyncio

async def task(name: str, delay: int, fail: bool = False) -> str:
    await asyncio.sleep(delay)
    if fail:
        raise ValueError(f"{name} 出错了")
    return f"{name} 完成"

async def main():
    # gather：等待所有任务，统一返回结果
    results = await asyncio.gather(
        task("A", 1),
        task("B", 2),
        task("C", 3),
        return_exceptions=True  # 把异常作为结果返回，而不是抛出
    )
    
    for i, r in enumerate(results):
        if isinstance(r, Exception):
            print(f"任务 {i} 失败：{r}")
        else:
            print(f"任务 {i} 成功：{r}")

asyncio.run(main())
# 任务 A 成功：A 完成
# 任务 B 成功：B 完成
# 任务 C 成功：C 完成

2-示例2

import asyncio
async def task(name: str, fail: bool = False):
    await asyncio.sleep(1)
    if fail:
        raise ValueError(f"{name} 失败")
    return f"{name} 成功"

async def main():
    tasks = [
        task("A", False),
        task("B", True),   # 这个会失败
        task("C", False),
    ]
    
    # 方式1：return_exceptions=True（推荐）
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for r in results:
        if isinstance(r, Exception):
            print(f"❌ {r}")
        else:
            print(f"✅ {r}")

asyncio.run(main())

示例3

import asyncio
async def safe_task(name: str) -> str:
    try:
        await asyncio.sleep(1)
        if name == "B":
            raise ValueError("B 出错了")
        return f"{name} 成功"
    except Exception as e:
        return f"{name} 失败：{e}"

async def main():
    results = await asyncio.gather(
        safe_task("A"),
        safe_task("B"),
        safe_task("C"),
    )
    print(results)

asyncio.run(main())

示例4-Semaphore —— 控制并发数

import asyncio
import random

class RateLimiter:
    def __init__(self, max_concurrent: int = 3):
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def request(self, url: str) -> str:
        async with self.semaphore:  # 获取信号量
            print(f"⬇️ 请求：{url}（并发中）")
            delay = random.uniform(0.5, 1.5)
            await asyncio.sleep(delay)
            return f"✅ {url} 完成"

async def main():
    limiter = RateLimiter(max_concurrent=3)
    
    # 同时启动 10 个任务，但最多 3 个并发
    tasks = [limiter.request(f"http://api.com/{i}") for i in range(10)]
    results = await asyncio.gather(*tasks)
    
    for r in results:
        print(r)

asyncio.run(main())

示例5- 天气查询 —

# 导入异步IO核心库，用于实现并发网络请求
import asyncio
# 随机数库，模拟请求延迟、随机报错、随机温度
import random
# 时间模块，统计程序整体运行耗时
import time
# 数据类装饰器，快速定义存储天气信息的数据模型
from dataclasses import dataclass
# 类型注解：列表、可选类型（可为None）
from typing import List, Optional

# 天气数据数据类：统一存储每个城市的查询结果
@dataclass
class WeatherData:
    # 城市名称
    city: str
    # 城市温度，请求失败时填0
    temp: int
    # 错误信息，查询正常则为None，可选字段默认None
    error: Optional[str] = None

# 异步天气请求API封装类
class AsyncWeatherAPI:
    def __init__(self, max_concurrent: int = 3, timeout: float = 2.0):
        """
        构造函数初始化请求限制与超时配置
        :param max_concurrent: 最大并发请求数，信号量控制限流
        :param timeout: 单次请求基础超时阈值
        """
        # 异步信号量：控制最大并发，防止同时发起过多请求触发接口限流
        self.semaphore = asyncio.Semaphore(max_concurrent)
        # 基础超时时间，批量总超时会基于该值放大
        self.timeout = timeout
    
    async def fetch_one(self, city: str) -> WeatherData:
        """
        获取单个城市天气（带并发限流保护）
        :param city: 需要查询的城市名
        :return: WeatherData 对象，包含温度或错误信息
        """
        # async with 自动获取/释放信号量，保证同一时间并发不超过上限
        async with self.semaphore:
            try:
                # 模拟网络请求延迟：0.3~1.5秒随机耗时
                delay = random.uniform(0.3, 1.5)
                # 异步休眠，不阻塞其他协程执行
                await asyncio.sleep(delay)
                
                # 10%概率模拟接口异常报错
                if random.random() < 0.1:
                    raise ValueError("API 错误")
                
                # 请求成功：返回随机15~35度的天气数据，无错误
                return WeatherData(city=city, temp=random.randint(15, 35))
            except Exception as e:
                # 捕获所有异常，返回带错误标记的空温度数据
                return WeatherData(city=city, temp=0, error=str(e))
    
    async def fetch_batch(self, cities: list[str]) -> list[WeatherData]:
        """
        批量查询多个城市天气，全局总超时控制
        :param cities: 待查询城市列表
        :return: 所有城市的天气结果列表
        """
        try:
            # asyncio.wait_for：给整批协程设置总超时时间
            # asyncio.gather：并发执行所有单个城市查询协程，收集全部结果
            return await asyncio.wait_for(
                asyncio.gather(*[self.fetch_one(c) for c in cities]),
                timeout=self.timeout * 2  # 批量总超时设为基础超时的2倍
            )
        except TimeoutError:
            # 整批请求超时：所有城市统一标记"整体超时"错误
            return [WeatherData(city=c, temp=0, error="整体超时") for c in cities]

# 程序主异步入口函数
async def main():
    # 实例化天气API：最大并发5，单次基础超时2秒
    api = AsyncWeatherAPI(max_concurrent=5, timeout=2.0)
    # 待查询城市列表
    cities = ["北京", "上海", "西安", "广州", "成都", "武汉", "南京", "杭州"]
    
    print(f"🌤️ 开始查询 {len(cities)} 个城市...")
    # 记录程序开始时间，用于计算总耗时
    start = time.time()
    
    # 批量异步查询所有城市天气
    results = await api.fetch_batch(cities)
    
    # 遍历打印每个城市的查询结果
    for data in results:
        if data.error:
            # 存在错误信息，打印失败日志
            print(f"  ❌ {data.city}：{data.error}")
        else:
            # 查询成功，打印城市温度
            print(f"  ✅ {data.city}：{data.temp}°C")
    
    # 计算并打印整体执行耗时，保留2位小数
    print(f"⏱️ 耗时：{time.time() - start:.2f}秒")

# 异步程序固定启动入口，运行主协程
asyncio.run(main())