一文搞懂Python asyncio异步编程的核心原理与实战技巧

import asyncio

async def say_hello(name: str, delay: int):
    await asyncio.sleep(delay)
    print(f"Hello, {name}! (after {delay}s)")
    return f"greeted {name}"

async def main():
    result = await say_hello("World", 2)
    print(f"Result: {result}")

asyncio.run(main())

import asyncio
import time

async def fetch_data(url: str, delay: int) -> dict:
    print(f"开始请求: {url}")
    await asyncio.sleep(delay)
    return {"url": url, "status": 200, "delay": delay}

async def main():
    start = time.time()
    urls = [
        ("https://api.example.com/users", 2),
        ("https://api.example.com/posts", 1),
        ("https://api.example.com/comments", 3),
    ]
    tasks = [fetch_data(url, delay) for url, delay in urls]
    results = await asyncio.gather(*tasks)

    elapsed = time.time() - start
    print(f"总耗时: {elapsed:.1f}s（而非串行的6s）")
    for r in results:
        print(f"  {r['url']} -> {r['status']}")

asyncio.run(main())

async def main_taskgroup():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_data("https://api.a.com", 2))
        task2 = tg.create_task(fetch_data("https://api.b.com", 1))
    print(task1.result(), task2.result())

import asyncio

class AsyncConnectionPool:
    def __init__(self, max_size: int = 5):
        self.max_size = max_size
        self._pool = asyncio.Queue(maxsize=max_size)

    async def connect(self):
        for i in range(self.max_size):
            await self._pool.put(f"conn_{i}")

    async def __aenter__(self):
        self.conn = await self._pool.get()
        print(f"获取连接: {self.conn}")
        return self.conn

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self._pool.put(self.conn)
        print(f"归还连接: {self.conn}")

async def main():
    pool = AsyncConnectionPool(max_size=3)
    await pool.connect()
    async with pool as conn:
        await asyncio.sleep(0.5)
        print(f"使用 {conn} 执行查询")

asyncio.run(main())

import asyncio

async def async_range(n: int):
    for i in range(n):
        await asyncio.sleep(0.1)
        yield i

async def paginate_api(page_size: int = 10, total_pages: int = 5):
    for page in range(1, total_pages + 1):
        await asyncio.sleep(0.3)
        data = [f"item_{page}_{i}" for i in range(page_size)]
        yield {"page": page, "data": data}

async def main():
    async for num in async_range(5):
        print(f"收到数据: {num}")
    async for page_result in paginate_api():
        print(f"第{page_result['page']}页: {len(page_result['data'])}条记录")

asyncio.run(main())

import asyncio
import aiohttp

async def fetch_with_limit(sem, url, session):
    async with sem:
        print(f"请求: {url}")
        async with session.get(url) as resp:
            return await resp.text()

async def main():
    sem = asyncio.Semaphore(5)
    urls = [f"https://httpbin.org/delay/{i % 3}" for i in range(20)]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_limit(sem, url, session) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        print(f"完成 {len(results)} 个请求")

async def slow_operation():
    await asyncio.sleep(100)
    return "done"

async def main_timeout():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=3.0)
    except asyncio.TimeoutError:
        print("操作超时，已取消")

asyncio.run(main_timeout())