Python异步编程从入门到精通

Python异步编程从入门到精通

Python的异步编程（asyncio）是处理I/O密集型任务的利器，特别是在网络请求、数据库查询和文件操作等场景下，可以显著提升程序性能。本文将系统介绍Python异步编程的核心概念和实战技巧。

一、协程基础

import asyncio

# 定义协程
async def fetch_data(url):
    print(f"开始请求: {url}")
    await asyncio.sleep(1)  # 模拟I/O操作
    print(f"请求完成: {url}")
    return {"url": url, "data": "响应内容"}

# 运行协程
result = asyncio.run(fetch_data("https://api.example.com"))

# 并发执行多个协程
async def main():
    urls = ["url1", "url2", "url3"]
    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks)
    return results

asyncio.run(main())

二、asyncio核心组件

# Task - 并发执行协程
async def main():
    task = asyncio.create_task(fetch_data("url1"))
    # 做其他事情
    result = await task

# asyncio.gather - 等待多个协程
results = await asyncio.gather(
    fetch_data("url1"),
    fetch_data("url2"),
    fetch_data("url3"),
    return_exceptions=True  # 异常不中断其他任务
)

# asyncio.wait - 更细粒度的控制
done, pending = await asyncio.wait(
    tasks,
    timeout=5.0,
    return_when=asyncio.FIRST_COMPLETED
)

# asyncio.Timeout - 超时控制
async with asyncio.timeout(5.0):
    await long_running_task()

三、异步HTTP请求

import aiohttp

async def fetch_json(session, url):
    async with session.get(url) as response:
        return await response.json()

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

# 限制并发数
semaphore = asyncio.Semaphore(10)

async def fetch_limited(session, url):
    async with semaphore:
        return await fetch_json(session, url)

四、异步数据库操作

import aiomysql

async def get_users():
    async with aiomysql.create_pool(
        host='localhost', port=3306,
        user='root', password='',
        db='myapp', charset='utf8mb4'
    ) as pool:
        async with pool.acquire() as conn:
            async with conn.cursor() as cur:
                await cur.execute("SELECT * FROM users")
                return await cur.fetchall()

# 异步ORM - SQLAlchemy 2.0
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

engine = create_async_engine("sqlite+aiosqlite:///db.sqlite3")
async with AsyncSession(engine) as session:
    result = await session.execute(select(User))
    users = result.scalars().all()

五、异步与同步的桥接

# 在异步中调用同步代码
import concurrent.futures

def cpu_intensive_task(n):
    return sum(i * i for i in range(n))

async def main():
    loop = asyncio.get_event_loop()
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(
            pool, cpu_intensive_task, 1000000
        )
    print(f"计算结果: {result}")

Python异步编程虽然有一定的学习曲线，但在I/O密集型场景下效果显著。建议从简单的网络请求开始，逐步掌握异步编程的思维方式，在合适的场景中发挥它的威力。