Python asyncio 极速异步编程完全指南：从协程到高并发服务

import requests
import time

def fetch_sync(urls):
    results = []
    for url in urls:
        resp = requests.get(url)      # 阻塞！CPU 干等着网络响应
        results.append(resp.text)
    return results

urls = [f"https://httpbin.org/get?i={i}" for i in range(10)]
start = time.perf_counter()
fetch_sync(urls)
print(f"同步耗时: {time.perf_counter() - start:.2f}s")
# → 约 15-20 秒（100个网页则 150-200 秒！）

import asyncio
import aiohttp
import time

async def fetch_one(session, url):
    async with session.get(url) as resp:
        return await resp.text()

async def fetch_async(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks)

urls = [f"https://httpbin.org/get?i={i}" for i in range(10)]
start = time.perf_counter()
asyncio.run(fetch_async(urls))
print(f"异步耗时: {time.perf_counter() - start:.2f}s")
# → 约 1-2 秒（100 个网页约 5-10 秒）

import asyncio

async def hello():
    print("开始")
    await asyncio.sleep(1)   # 暂停！把控制权交还事件循环
    print("1秒后")

# hello() 被调用时，不会立即执行，而是返回一个协程对象
coro = hello()
print(type(coro))   # <class 'coroutine'>

# 需要事件循环来驱动它
asyncio.run(coro)   # Python 3.7+ 推荐写法

import asyncio

async def main():
    print("开始事件循环")
    await asyncio.sleep(0.5)
    print("结束")

# 获取当前事件循环（低层 API，通常不需要手动使用）
loop = asyncio.get_event_loop()

# Python 3.7+ 高层 API：自动创建、运行、关闭
asyncio.run(main())

import asyncio

async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)

async def main():
    # task1 = say_after(1, "hello")    # ❌ 这只是协程对象，没被调度
    
    task1 = asyncio.create_task(say_after(1, "你好"))
    task2 = asyncio.create_task(say_after(2, "世界"))
    
    print(f"创建于 {asyncio.get_event_loop().time():.2f}")
    
    await task1   # 等待任务1完成
    await task2   # 等待任务2完成

asyncio.run(main())
# 总耗时约 2 秒（不是 3 秒，因为它们是并发运行的）

import asyncio

# async def 定义协程
async def fetch_data():
    # await 只能出现在 async def 内部
    await asyncio.sleep(1)   # 暂停，让出 CPU
    return "数据"

async def main():
    # await 表达式的结果是协程的返回值
    result = await fetch_data()
    print(result)   # 数据

asyncio.run(main())

async def demo():
    # ✅ 合法
    await asyncio.sleep(1)           # 协程
    await asyncio.create_task(coro)  # Task
    await some_future                # Future
    
    # ❌ 不合法
    # await 123           # 错误！
    # await [coro1, coro2] # list 不是 Awaitable

import asyncio

async def async_func():
    await asyncio.sleep(1)
    return "done"

# 方式 1：asyncio.run()（最顶层入口）
asyncio.run(async_func())

# 方式 2：在已有事件循环中获取结果
async def main():
    result = await async_func()
    return result

# 方式 3：同步代码中运行协程（注意：不要在已有 loop 中调用！）
# result = asyncio.run(async_func())

import asyncio

async def fetch_url(url):
    await asyncio.sleep(1)  # 模拟网络请求
    return f"Result of {url}"

async def main():
    urls = ["url1", "url2", "url3", "url4", "url5"]
    
    # gather：并发运行所有协程，等待全部完成
    # 约 1 秒完成（不是 5 秒！）
    results = await asyncio.gather(
        *[fetch_url(url) for url in urls]
    )
    
    for r in results:
        print(r)

asyncio.run(main())

async def may_fail(x):
    if x == 2:
        raise ValueError(f"Error at {x}")
    await asyncio.sleep(0.1)
    return x * x

async def main():
    # 默认：一个异常，全部取消
    try:
        results = await asyncio.gather(
            may_fail(1), may_fail(2), may_fail(3)
        )
    except ValueError as e:
        print(f"捕获异常: {e}")
    
    # return_exceptions=True：返回异常而非抛出
    results = await asyncio.gather(
        may_fail(1), may_fail(2), may_fail(3),
        return_exceptions=True
    )
    print(results)   # [1, ValueError('Error at 2'), 9]

asyncio.run(main())

import asyncio

async def slow_task(name, delay):
    await asyncio.sleep(delay)
    return f"{name} completed"

async def main():
    tasks = [
        asyncio.create_task(slow_task("A", 2)),
        asyncio.create_task(slow_task("B", 1)),
        asyncio.create_task(slow_task("C", 3)),
    ]
    
    # return_when=ALL_COMPLETED（默认）
    # return_when=FIRST_COMPLETED —— 任意一个完成就返回
    done, pending = await asyncio.wait(
        tasks,
        return_when=asyncio.FIRST_COMPLETED
    )
    
    for task in done:
        print(f"已完成: {task.result()}")
    
    # 别忘了取消剩余任务
    for task in pending:
        task.cancel()

asyncio.run(main())

import asyncio

async def slow_operation():
    await asyncio.sleep(10)
    return "Done"

async def main():
    # 方式 1：wait_for（超时抛异常）
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
    except asyncio.TimeoutError:
        print("操作超时！")
    
    # 方式 2：timeout 上下文管理器（Python 3.11+）
    try:
        async with asyncio.timeout(2.0):
            result = await slow_operation()
    except TimeoutError:
        print("操作超时！")

asyncio.run(main())

import asyncio

# 同时最多 3 个并发
sem = asyncio.Semaphore(3)

async def fetch_with_limit(url):
    async with sem:   # 获取信号量，超限时等待
        print(f"开始下载 {url}")
        await asyncio.sleep(1)
        print(f"完成下载 {url}")
        return f"Content of {url}"

async def main():
    urls = [f"https://example.com/{i}" for i in range(10)]
    results = await asyncio.gather(*[fetch_with_limit(url) for url in urls])
    print(f"共下载 {len(results)} 个页面")

asyncio.run(main())

import asyncio

async def producer(queue, n):
    for i in range(n):
        await asyncio.sleep(0.1)
        await queue.put(i)
        print(f"生产: {i}")
    await queue.put(None)   # 哨兵，表示结束

async def consumer(queue, name):
    while True:
        item = await queue.get()
        if item is None:
            break
        await asyncio.sleep(0.2)
        print(f"消费者{name} 处理: {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=5)
    
    prod_task = asyncio.create_task(producer(queue, 10))
    cons_tasks = [
        asyncio.create_task(consumer(queue, i))
        for i in range(2)   # 2 个消费者
    ]
    
    await prod_task
    await queue.join()   # 等待所有任务处理完
    
    for t in cons_tasks:
        t.cancel()

asyncio.run(main())

import asyncio

class AsyncConnection:
    """模拟一个需要异步打开和关闭的资源"""
    
    async def __aenter__(self):
        print("异步连接建立...")
        await asyncio.sleep(0.5)
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("异步连接关闭...")
        await asyncio.sleep(0.5)

async def main():
    async with AsyncConnection() as conn:
        print("使用连接中...")

asyncio.run(main())

import asyncio
import random

class AsyncPageFetcher:
    """异步分页获取数据"""
    
    def __init__(self, total_pages):
        self.total_pages = total_pages
    
    def __aiter__(self):
        self.current = 0
        return self
    
    async def __anext__(self):
        if self.current >= self.total_pages:
            raise StopAsyncIteration
        
        await asyncio.sleep(random.uniform(0.1, 0.3))
        self.current += 1
        return f"第 {self.current} 页数据"

async def main():
    fetcher = AsyncPageFetcher(5)
    async for page_data in fetcher:
        print(page_data)

asyncio.run(main())

import asyncio
import time

def cpu_intensive(n):
    """模拟 CPU 密集型同步操作"""
    count = 0
    for i in range(n):
        count += i ** 2
    return count

async def main():
    loop = asyncio.get_event_loop()
    
    # 在线程池中运行同步函数（不阻塞事件循环）
    result = await loop.run_in_executor(None, cpu_intensive, 10_000_000)
    print(f"计算结果: {result}")
    
    # 多个 CPU 密集型任务并行
    with concurrent.futures.ProcessPoolExecutor() as pool:
        tasks = [
            loop.run_in_executor(pool, cpu_intensive, n)
            for n in [5_000_000, 5_000_000, 5_000_000]
        ]
        results = await asyncio.gather(*tasks)
        print(f"并行结果: {results}")

import concurrent.futures
asyncio.run(main())

import asyncio
import aiohttp
import time
from urllib.parse import urljoin, urlparse

class AsyncCrawler:
    """异步爬虫：并发抓取网页并提取链接"""
    
    def __init__(self, base_url, max_concurrent=10, max_depth=2):
        self.base_url = base_url
        self.max_concurrent = max_concurrent
        self.max_depth = max_depth
        self.visited = set()
        self.results = []
    
    async def fetch(self, session, url, semaphore):
        """获取单个页面"""
        async with semaphore:
            try:
                async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                    if resp.status == 200:
                        return await resp.text()
            except Exception as e:
                print(f"抓取失败 {url}: {e}")
            return None
    
    async def crawl(self):
        """主爬取逻辑"""
        semaphore = asyncio.Semaphore(self.max_concurrent)
        
        async with aiohttp.ClientSession() as session:
            html = await self.fetch(session, self.base_url, semaphore)
            if html:
                self.results.append({
                    "url": self.base_url,
                    "length": len(html),
                    "status": "ok"
                })
        
        return self.results

async def main():
    urls = [
        "https://httpbin.org/get",
        "https://httpbin.org/ip",
        "https://httpbin.org/user-agent",
    ] * 5   # 15 个请求
    
    semaphore = asyncio.Semaphore(10)
    
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_one(session, url, semaphore)
            for url in urls
        ]
        
        start = time.perf_counter()
        results = await asyncio.gather(*tasks, return_exceptions=True)
        elapsed = time.perf_counter() - start
        
        success = sum(1 for r in results if isinstance(r, str))
        print(f"✅ 成功: {success}/{len(urls)}")
        print(f"⏱ 耗时: {elapsed:.2f}s")
        print(f"⚡ 平均: {elapsed/len(urls):.3f}s/请求")

async def fetch_one(session, url, sem):
    async with sem:
        async with session.get(url) as resp:
            return await resp.text()

asyncio.run(main())

from fastapi import FastAPI, HTTPException
import asyncio
import httpx

app = FastAPI(title="异步聚合 API")

# 异步 HTTP 客户端（比 requests 更适合 async 环境）
async_client = httpx.AsyncClient(timeout=10.0)

@app.get("/")
async def root():
    return {"message": "异步服务运行中"}

@app.get("/aggregate")
async def aggregate_data():
    """
    并发请求多个外部 API，聚合结果返回
    """
    urls = {
        "ip": "https://httpbin.org/ip",
        "user_agent": "https://httpbin.org/user-agent",
        "headers": "https://httpbin.org/headers",
    }
    
    # 并发请求所有 API
    tasks = {
        name: asyncio.create_task(async_client.get(url))
        for name, url in urls.items()
    }
    
    results = {}
    for name, task in tasks.items():
        try:
            resp = await task
            results[name] = resp.json()
        except Exception as e:
            results[name] = {"error": str(e)}
    
    return {
        "status": "success",
        "data": results
    }

@app.get("/slow")
async def slow_endpoint():
    """模拟耗时操作"""
    await asyncio.sleep(2)
    return {"message": "慢操作完成"}

@app.on_event("shutdown")
async def shutdown():
    await async_client.aclose()

# 启动：uvicorn main:app --reload

import asyncio
import asyncpg  # pip install asyncpg

# PostgreSQL 异步连接
async def demo_db():
    conn = await asyncpg.connect("postgresql://user:pass@localhost/db")
    
    # 执行查询
    rows = await conn.fetch("SELECT * FROM users WHERE active = $1", True)
    
    for row in rows:
        print(row["name"], row["email"])
    
    # 事务
    async with conn.transaction():
        await conn.execute(
            "INSERT INTO users (name, email) VALUES ($1, $2)",
            "Alice", "alice@example.com"
        )
    
    await conn.close()

# 连接池（生产环境推荐）
async def demo_pool():
    pool = await asyncpg.create_pool(
        "postgresql://user:pass@localhost/db",
        min_size=10,
        max_size=20
    )
    
    async with pool.acquire() as conn:
        row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", 1)
        print(row)
    
    await pool.close()

# asyncio.run(demo_pool())

# Tortoise ORM 示例
from tortoise import Tortoise, fields
from tortoise.models import Model

class User(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=100)
    email = fields.CharField(max_length=255, unique=True)
    created_at = fields.DatetimeField(auto_now_add=True)

async def init():
    await Tortoise.init(
        db_url="sqlite://db.sqlite3",
        modules={"models": ["__main__"]}
    )
    await Tortoise.generate_schemas()

async def create_user():
    user = await User.create(name="Alice", email="alice@example.com")
    print(f"创建用户: {user.id}")
    
    users = await User.filter(name__startswith="A").all()
    print(f"找到 {len(users)} 个用户")

# ❌ 错误！阻塞了整个事件循环
async def bad():
    requests.get("https://example.com")   # requests 是同步库
    time.sleep(1)                          # 阻塞睡眠

# ✅ 正确：使用异步库
async def good():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://example.com") as resp:
            text = await resp.text()
    await asyncio.sleep(1)

async def task():
    await asyncio.sleep(1)
    return "done"

async def main():
    result = task()       # ❌ result 是协程对象，不是 "done"！
    print(result)         # <coroutine object task at ...>
    
    result = await task()  # ✅ 正确
    print(result)          # done

# ❌ asyncio.run() 不能嵌套
async def main():
    asyncio.run(some_other())   # RuntimeError!

# ✅ 正确：用 await
async def main():
    await some_other()

async def failing_task():
    await asyncio.sleep(1)
    raise ValueError("出错了")

async def main():
    task = asyncio.create_task(failing_task())
    # 如果不 await task，异常可能被静默忽略或延迟抛到事件循环关闭时
    
    # ✅ 正确做法：
    try:
        await task
    except ValueError as e:
        print(f"捕获: {e}")
    
    # 或：
    task.add_done_callback(lambda t: print(f"任务结束: {t.exception()}"))

asyncio.run(main())

# 开启 asyncio 调试模式
asyncio.run(main(), debug=True)

# 设置慢协程警告阈值（默认 100ms）
import warnings
warnings.filterwarnings("default", category=RuntimeWarning)

# 查看事件循环中所有待处理任务
async def debug_tasks():
    tasks = asyncio.all_tasks()
    for t in tasks:
        print(f"任务: {t.get_name()}, 完成: {t.done()}")