asyncio 基础

import asyncio


async def greet(name: str) -> str:
    """协程函数：用 async def 定义"""
    await asyncio.sleep(1)  # 模拟异步 I/O 操作
    return f"Hello, {name}!"


# 调用协程函数，返回协程对象（不会执行函数体！）
coro = greet("Python")
print(type(coro))  # <class 'coroutine'>

# 必须通过 await 或 asyncio.run() 驱动执行
result = asyncio.run(greet("Python"))
print(result)  # Hello, Python!

async def fetch_data(url: str) -> dict:
    """模拟异步获取数据"""
    print(f"开始请求: {url}")
    await asyncio.sleep(2)  # 模拟网络延迟，暂停当前协程
    print(f"请求完成: {url}")
    return {"url": url, "status": 200}


async def main():
    # 串行执行：总耗时 ≈ 4s
    data1 = await fetch_data("https://api.example.com/users")
    data2 = await fetch_data("https://api.example.com/posts")
    print(f"获取到 {len([data1, data2])} 个结果")


asyncio.run(main())

async def compute() -> int:
    await asyncio.sleep(0.5)
    return 42


async def main():
    # 正确：await 获取返回值
    result = await compute()
    print(result)  # 42

    # 错误：直接调用只得到协程对象
    coro = compute()
    print(type(coro))  # <class 'coroutine'>
    # 必须 await 这个协程对象，否则会收到 RuntimeWarning
    result2 = await coro
    print(result2)  # 42


asyncio.run(main())

import asyncio


async def main():
    print("Hello from asyncio!")
    await asyncio.sleep(1)
    print("1 second later...")


# asyncio.run() 是启动异步程序的唯一推荐方式
asyncio.run(main())

import asyncio


async def task(name: str, seconds: int):
    print(f"{name} 开始")
    await asyncio.sleep(seconds)
    print(f"{name} 完成 (耗时 {seconds}s)")
    return f"{name} 的结果"


async def main():
    # 推荐方式：使用 asyncio.run() 启动
    # 以下代码在协程内部获取当前运行的事件循环
    loop = asyncio.get_running_loop()
    print(f"事件循环类型: {type(loop).__name__}")

    # 在协程内部安排回调
    loop.call_soon(lambda: print("回调执行！"))

    await asyncio.sleep(0.1)  # 给回调执行的机会


asyncio.run(main())

import asyncio
import time


async def fetch(url: str, delay: float) -> str:
    """模拟异步 HTTP 请求"""
    print(f"[{time.strftime('%H:%M:%S')}] 开始请求: {url}")
    await asyncio.sleep(delay)  # 模拟网络延迟
    print(f"[{time.strftime('%H:%M:%S')}] 完成请求: {url}")
    return f"响应来自 {url}"


async def main():
    start = time.time()

    # 串行执行：总耗时 ≈ 6s
    # result1 = await fetch("api/users", 2)
    # result2 = await fetch("api/posts", 1)
    # result3 = await fetch("api/comments", 3)

    # 并发执行：总耗时 ≈ 3s（取决于最慢的任务）
    task1 = asyncio.create_task(fetch("api/users", 2))
    task2 = asyncio.create_task(fetch("api/posts", 1))
    task3 = asyncio.create_task(fetch("api/comments", 3))

    # 等待所有 Task 完成
    result1 = await task1
    result2 = await task2
    result3 = await task3

    elapsed = time.time() - start
    print(f"总耗时: {elapsed:.1f}s")
    print(f"结果: {result1}, {result2}, {result3}")


asyncio.run(main())

[14:30:01] 开始请求: api/users
[14:30:01] 开始请求: api/posts
[14:30:01] 开始请求: api/comments
[14:30:02] 完成请求: api/posts
[14:30:03] 完成请求: api/users
[14:30:04] 完成请求: api/comments
总耗时: 3.0s

import asyncio


async def long_running():
    """模拟长时间运行的任务"""
    try:
        await asyncio.sleep(10)
        return "完成"
    except asyncio.CancelledError:
        print("任务被取消，执行清理...")
        raise  # 推荐重新抛出，让调用者知道任务被取消


async def main():
    # 创建 Task
    task = asyncio.create_task(long_running())

    # 检查状态
    print(f"完成? {task.done()}")   # False
    print(f"已取消? {task.cancelled()}")  # False

    # 等待一小段时间后取消
    await asyncio.sleep(1)
    task.cancel()  # 请求取消

    try:
        await task  # 等待取消完成
    except asyncio.CancelledError:
        print("主协程: 任务已被取消")

    print(f"完成? {task.done()}")   # True
    print(f"已取消? {task.cancelled()}")  # True


asyncio.run(main())

import asyncio
import time


async def fetch_api(name: str, delay: float) -> dict:
    """模拟 API 请求"""
    await asyncio.sleep(delay)
    return {"api": name, "delay": delay, "data": f"{name} 的数据"}


async def main():
    start = time.time()

    # gather 并发执行，结果按传入顺序排列
    results = await asyncio.gather(
        fetch_api("用户服务", 2),
        fetch_api("订单服务", 1),
        fetch_api("商品服务", 3),
    )

    elapsed = time.time() - start
    print(f"总耗时: {elapsed:.1f}s")  # ≈ 3s
    for r in results:
        print(f"  {r['api']}: delay={r['delay']}s")


asyncio.run(main())

import asyncio


async def success_task():
    await asyncio.sleep(1)
    return "成功"


async def fail_task():
    await asyncio.sleep(0.5)
    raise ValueError("出错了！")


async def main():
    # 默认行为：遇到异常立即抛出
    try:
        results = await asyncio.gather(success_task(), fail_task())
    except ValueError as e:
        print(f"默认模式捕获异常: {e}")

    # return_exceptions=True：异常作为结果返回
    results = await asyncio.gather(
        success_task(), fail_task(), return_exceptions=True
    )
    for i, r in enumerate(results):
        if isinstance(r, Exception):
            print(f"任务 {i}: 异常 - {r}")
        else:
            print(f"任务 {i}: 成功 - {r}")

    # 输出:
    # 任务 0: 成功 - 成功
    # 任务 1: 异常 - 出错了！


asyncio.run(main())

import asyncio
import random


async def random_task(task_id: int):
    """随机耗时的任务"""
    delay = random.uniform(1, 5)
    await asyncio.sleep(delay)
    return f"任务 {task_id} 完成 (耗时 {delay:.1f}s)"


async def main():
    tasks = [asyncio.create_task(random_task(i)) for i in range(5)]

    # FIRST_COMPLETED: 任何一个完成就返回
    done, pending = await asyncio.wait(
        tasks, return_when=asyncio.FIRST_COMPLETED
    )
    print(f"第一个完成的: {done.pop().result()}")
    # 取消剩余任务
    for t in pending:
        t.cancel()
    await asyncio.gather(*pending, return_exceptions=True)

    # FIRST_EXCEPTION: 第一个异常或全部完成
    # ALL_COMPLETED: 等待全部完成（默认行为，等同于 gather）


asyncio.run(main())

import asyncio


async def main():
    # wait 的超时控制
    tasks = [asyncio.create_task(asyncio.sleep(i)) for i in range(1, 6)]

    done, pending = await asyncio.wait(tasks, timeout=2.5)
    print(f"超时前完成: {len(done)} 个")  # 2 个 (1s, 2s)
    print(f"超时未完成: {len(pending)} 个")  # 3 个 (3s, 4s, 5s)

    # 清理未完成的任务
    for t in pending:
        t.cancel()
    await asyncio.gather(*pending, return_exceptions=True)


asyncio.run(main())

import asyncio


async def fetch_data(name: str, delay: float):
    await asyncio.sleep(delay)
    if name == "失败服务":
        raise ConnectionError(f"{name} 连接失败")
    return f"{name} 的数据"


async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(fetch_data("服务A", 1))
            t2 = tg.create_task(fetch_data("失败服务", 2))
            t3 = tg.create_task(fetch_data("服务C", 3))
    except* ConnectionError as eg:  # ExceptionGroup 语法 (Python 3.11+)
        print(f"捕获到异常组: {eg.exceptions}")

    # t1 和 t3 也会被自动取消，因为 t2 抛出了异常


asyncio.run(main())

import asyncio


class AsyncDatabaseConnection:
    """模拟异步数据库连接"""

    def __init__(self, db_url: str):
        self.db_url = db_url
        self._connected = False

    async def __aenter__(self):
        print(f"连接数据库: {self.db_url}")
        await asyncio.sleep(0.5)  # 模拟异步连接
        self._connected = True
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print(f"关闭数据库连接: {self.db_url}")
        await asyncio.sleep(0.2)  # 模拟异步关闭
        self._connected = False
        # 返回 False 表示不吞掉异常，True 表示吞掉异常
        return False

    async def execute(self, query: str):
        if not self._connected:
            raise RuntimeError("数据库未连接")
        await asyncio.sleep(0.1)  # 模拟异步查询
        return f"查询结果: {query}"


async def main():
    async with AsyncDatabaseConnection("postgresql://localhost/mydb") as db:
        result = await db.execute("SELECT * FROM users")
        print(result)
    # 即使发生异常，连接也会被正确关闭


asyncio.run(main())

import asyncio
from contextlib import asynccontextmanager


@asynccontextmanager
async def async_timer(name: str):
    """异步计时器：自动测量代码块执行时间"""
    start = asyncio.get_event_loop().time()
    print(f"[{name}] 开始执行")
    try:
        yield  # 在这里执行 with 块中的代码
    finally:
        elapsed = asyncio.get_event_loop().time() - start
        print(f"[{name}] 执行完成，耗时 {elapsed:.3f}s")


async def main():
    async with async_timer("数据处理"):
        await asyncio.sleep(1.5)
        print("处理中...")


asyncio.run(main())

import asyncio
from contextlib import asynccontextmanager


class AsyncPool:
    """异步连接池示例"""

    def __init__(self, max_size: int = 10):
        self._max_size = max_size
        self._semaphore = asyncio.Semaphore(max_size)
        self._connections = []

    @asynccontextmanager
    async def acquire(self):
        """获取一个连接，用完自动归还"""
        await self._semaphore.acquire()
        conn = f"conn_{len(self._connections)}"
        self._connections.append(conn)
        try:
            yield conn
        finally:
            self._connections.remove(conn)
            self._semaphore.release()


async def main():
    pool = AsyncPool(max_size=3)

    async with pool.acquire() as conn:
        print(f"使用连接: {conn}")
        await asyncio.sleep(1)

    # 连接已自动归还


asyncio.run(main())

import asyncio


class AsyncPageFetcher:
    """异步分页数据获取器"""

    def __init__(self, total_items: int, page_size: int = 10):
        self.total_items = total_items
        self.page_size = page_size
        self._current_page = 0

    def __aiter__(self):
        return self

    async def __anext__(self) -> list[dict]:
        if self._current_page * self.page_size >= self.total_items:
            raise StopAsyncIteration

        # 模拟异步 API 请求获取一页数据
        start = self._current_page * self.page_size
        end = min(start + self.page_size, self.total_items)
        await asyncio.sleep(0.5)  # 模拟网络延迟

        page_data = [
            {"id": i, "name": f"item_{i}"} for i in range(start, end)
        ]
        self._current_page += 1
        print(f"获取第 {self._current_page} 页，共 {len(page_data)} 条")
        return page_data


async def main():
    # 异步遍历分页数据
    all_items = []
    async for page in AsyncPageFetcher(total_items=25, page_size=10):
        all_items.extend(page)

    print(f"总共获取 {len(all_items)} 条数据")


asyncio.run(main())

import asyncio


async def async_range(start: int, end: int, delay: float = 0.1):
    """异步生成器：每隔 delay 秒产出一个值"""
    for i in range(start, end):
        await asyncio.sleep(delay)
        yield i


async def stream_lines(file_path: str):
    """异步逐行读取文件（模拟）"""
    lines = [f"第 {i} 行内容" for i in range(1, 6)]
    for line in lines:
        await asyncio.sleep(0.2)  # 模拟异步 I/O
        yield line


async def main():
    # 使用 async for 遍历异步生成器
    print("异步范围:")
    async for num in async_range(1, 5):
        print(f"  {num}")

    print("\n异步文件流:")
    async for line in stream_lines("/tmp/data.txt"):
        print(f"  {line}")


asyncio.run(main())

import asyncio


async def async_range(start: int, end: int, delay: float = 0.1):
    for i in range(start, end):
        await asyncio.sleep(delay)
        yield i


async def main():
    # 异步列表推导式（Python 3.6+）
    result = [i async for i in async_range(1, 6)]
    print(result)  # [1, 2, 3, 4, 5]

    # 带条件的异步推导式
    evens = [i async for i in async_range(1, 11) if i % 2 == 0]
    print(evens)  # [2, 4, 6, 8, 10]

    # 异步集合推导式
    unique = {i % 3 async for i in async_range(1, 7)}
    print(unique)  # {0, 1, 2}

    # 异步字典推导式
    squares = {i: i**2 async for i in async_range(1, 5)}
    print(squares)  # {1: 1, 2: 4, 3: 9, 4: 16}


asyncio.run(main())

import asyncio
import time
from typing import Any


# 模拟异步 HTTP 请求函数（实际项目中替换为 aiohttp 或 httpx）
async def async_get(url: str, delay: float = 1.0) -> dict[str, Any]:
    """模拟异步 GET 请求"""
    await asyncio.sleep(delay)  # 模拟网络延迟
    return {"url": url, "status": 200, "data": f"来自 {url} 的响应"}


async def fetch_user(user_id: int) -> dict:
    """获取用户信息"""
    return await async_get(f"https://api.example.com/users/{user_id}", delay=0.5)


async def fetch_user_posts(user_id: int) -> list:
    """获取用户文章"""
    data = await async_get(f"https://api.example.com/users/{user_id}/posts", delay=0.8)
    return data.get("data", [])


async def fetch_user_with_posts(user_id: int) -> dict:
    """并发获取用户信息及其文章"""
    # 两个请求并发执行
    user, posts = await asyncio.gather(
        fetch_user(user_id),
        fetch_user_posts(user_id),
    )
    return {**user, "posts": posts}


async def batch_fetch_users(user_ids: list[int], max_concurrent: int = 5) -> list[dict]:
    """批量获取用户信息，控制最大并发数"""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def limited_fetch(uid: int):
        async with semaphore:
            return await fetch_user_with_posts(uid)

    tasks = [limited_fetch(uid) for uid in user_ids]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # 分离成功和失败的结果
    success = []
    failures = []
    for uid, result in zip(user_ids, results):
        if isinstance(result, Exception):
            failures.append({"user_id": uid, "error": str(result)})
        else:
            success.append(result)

    print(f"成功: {len(success)}, 失败: {len(failures)}")
    return success


async def main():
    start = time.time()

    # 批量获取 10 个用户的信息，最大并发 5
    user_ids = list(range(1, 11))
    results = await batch_fetch_users(user_ids, max_concurrent=5)

    elapsed = time.time() - start
    print(f"获取 {len(results)} 个用户，总耗时: {elapsed:.1f}s")
    # 并发执行，总耗时远小于串行执行


asyncio.run(main())

import asyncio
import time
from typing import AsyncIterator


# 模拟异步文件读取（实际项目中使用 aiofiles）
async def async_read_lines(file_path: str) -> AsyncIterator[str]:
    """异步逐行读取文件"""
    # 模拟文件内容
    lines = [
        "name,age,score",
        "Alice,25,92",
        "Bob,30,85",
        "Charlie,28,78",
        "Diana,22,95",
        "Eve,35,88",
    ]
    for line in lines:
        await asyncio.sleep(0.1)  # 模拟异步 I/O
        yield line


async def transform_line(line: str) -> str | None:
    """异步转换每一行数据"""
    if line.startswith("name,"):
        return None  # 跳过表头

    await asyncio.sleep(0.05)  # 模拟异步处理（如调用 API 补充数据）

    parts = line.split(",")
    name, age, score = parts[0], parts[1], parts[2]
    grade = "A" if int(score) >= 90 else "B" if int(score) >= 80 else "C"
    return f"{name}|{age}|{score}|{grade}"


async def async_write_line(file_path: str, line: str) -> None:
    """异步写入一行数据"""
    await asyncio.sleep(0.05)  # 模拟异步 I/O
    # 实际项目中: async with aiofiles.open(file_path, 'a') as f: await f.write(line)


async def process_pipeline(input_path: str, output_path: str) -> dict:
    """异步数据处理管道：读取 -> 转换 -> 写入"""
    processed = 0
    skipped = 0
    errors = 0

    async for line in async_read_lines(input_path):
        try:
            result = await transform_line(line)
            if result is None:
                skipped += 1
                continue
            await async_write_line(output_path, result)
            processed += 1
        except Exception as e:
            print(f"处理失败: {line}, 错误: {e}")
            errors += 1

    return {"processed": processed, "skipped": skipped, "errors": errors}


async def process_pipeline_concurrent(
    input_path: str, output_path: str, batch_size: int = 3
) -> dict:
    """并发版数据处理管道：批量转换，提高吞吐量"""
    processed = 0
    skipped = 0
    errors = 0

    batch = []
    async for line in async_read_lines(input_path):
        batch.append(line)
        if len(batch) >= batch_size:
            # 并发处理一批数据
            results = await asyncio.gather(
                *[transform_line(l) for l in batch],
                return_exceptions=True,
            )
            for result in results:
                if isinstance(result, Exception):
                    errors += 1
                elif result is None:
                    skipped += 1
                else:
                    await async_write_line(output_path, result)
                    processed += 1
            batch = []

    # 处理剩余数据
    if batch:
        results = await asyncio.gather(
            *[transform_line(l) for l in batch], return_exceptions=True
        )
        for result in results:
            if isinstance(result, Exception):
                errors += 1
            elif result is None:
                skipped += 1
            else:
                await async_write_line(output_path, result)
                processed += 1

    return {"processed": processed, "skipped": skipped, "errors": errors}


async def main():
    start = time.time()

    # 串行管道
    stats = await process_pipeline("input.csv", "output.csv")
    elapsed1 = time.time() - start
    print(f"串行管道: {stats}, 耗时 {elapsed1:.2f}s")

    # 并发管道
    start = time.time()
    stats = await process_pipeline_concurrent("input.csv", "output.csv")
    elapsed2 = time.time() - start
    print(f"并发管道: {stats}, 耗时 {elapsed2:.2f}s")


asyncio.run(main())

import asyncio


async def debug_example():
    # 1. 开启 asyncio 调试模式
    # 方式一：asyncio.run 参数
    # asyncio.run(main(), debug=True)

    # 方式二：环境变量
    # PYTHONASYNCIODEBUG=1 python script.py

    # 2. 检测未 await 的协程
    # Python 3.11+ 会在协程未被 await 时发出 RuntimeWarning

    # 3. 检测阻塞事件循环的操作
    # debug=True 时，事件循环会记录执行时间超过 100ms 的操作
    pass


# 4. 使用 asyncio 的内置日志
import logging
logging.getLogger("asyncio").setLevel(logging.DEBUG)