Python 中的异步编程：从基础到实战

 

一、异步编程的核心概念

 

1. 同步 vs 异步

 

- 同步编程：代码按顺序逐行执行，前一个任务完成才能执行下一个（如传统HTTP请求）

- 异步编程：允许在等待I/O操作（如网络请求、文件读写）时切换执行其他任务，提高资源利用率

 

2. 核心组件

 

- 协程(Coroutine)：可暂停/恢复的函数，使用

"async def"定义

- 事件循环(Event Loop)：调度协程执行的中央控制器

- Future/Task：表示异步操作最终结果的对象

 

二、基础语法与实现

 

1. 协程定义与调用

 

import asyncio

 

# 定义协程函数

async def say_hello(name):

    print(f"准备问候 {name}...")

    await asyncio.sleep(1) # 模拟I/O操作

    print(f"Hello, {name}!")

 

# 调用协程

async def main():

    await say_hello("World") # 正确方式：在协程内await

 

# 运行事件循环

asyncio.run(main()) # Python 3.7+

 

2. 并发执行多个协程

 

async def fetch_data(id, delay):

    print(f"任务 {id} 开始获取数据")

    await asyncio.sleep(delay)

    print(f"任务 {id} 完成，耗时 {delay}秒")

    return f"数据-{id}"

 

async def main():

    # 方式1：asyncio.gather (推荐)

    results = await asyncio.gather(

        fetch_data(1, 2),

        fetch_data(2, 1),

        fetch_data(3, 3)

    )

    print("所有结果:", results)

 

    # 方式2：asyncio.create_task (更灵活)

    task1 = asyncio.create_task(fetch_data(1, 2))

    task2 = asyncio.create_task(fetch_data(2, 1))

    await task1

    await task2

 

asyncio.run(main())

 

三、高级应用场景

 

1. 超时控制

 

async def long_running_task():

    try:

        async with asyncio.timeout(2): # Python 3.11+

            await asyncio.sleep(3)

            print("任务完成")

    except TimeoutError:

        print("任务超时被取消")

 

asyncio.run(long_running_task())

 

2. 任务取消

 

async def cancellable_task():

    try:

        while True:

            print("任务运行中...")

            await asyncio.sleep(1)

    except asyncio.CancelledError:

        print("任务被取消")

        raise

 

async def main():

    task = asyncio.create_task(cancellable_task())

    await asyncio.sleep(3)

    task.cancel()

    try:

        await task

    except asyncio.CancelledError:

        print("主程序捕获到任务取消")

 

asyncio.run(main())

 

3. 异步上下文管理器

 

class AsyncResource:

    async def __aenter__(self):

        print("获取资源")

        await asyncio.sleep(1)

        return self

    

    async def __aexit__(self, exc_type, exc, tb):

        print("释放资源")

        await asyncio.sleep(0.5)

 

async def use_resource():

    async with AsyncResource() as resource:

        print("正在使用资源...")

        await asyncio.sleep(1)

 

asyncio.run(use_resource())

 

四、实际项目案例

 

1. 异步Web爬虫

 

import aiohttp

 

async def fetch_url(session, url):

    async with session.get(url) as response:

        return await response.text()

 

async def crawl_website(urls):

    async with aiohttp.ClientSession() as session:

        tasks = [fetch_url(session, url) for url in urls]

        results = await asyncio.gather(*tasks)

        for url, content in zip(urls, results):

            print(f"{url} 获取到 {len(content)} 字节")

 

urls = [

    "https://www.example.com",

    "https://www.python.org",

    "https://www.github.com"

]

asyncio.run(crawl_website(urls))

 

2. 异步数据库操作

 

import asyncpg

 

async def query_database():

    conn = await asyncpg.connect(

        user='user', password='pass',

        database='db', host='localhost'

    )

    try:

        # 并发执行多个查询

        user_task = conn.fetch('SELECT * FROM users WHERE id = $1', 1)

        product_task = conn.fetch('SELECT * FROM products LIMIT 5')

        

        user, products = await asyncio.gather(user_task, product_task)

        print(f"用户: {user}, 商品数: {len(products)}")

    finally:

        await conn.close()

 

asyncio.run(query_database())

 

五、调试与性能优化

 

1. 常见问题排查

 

- 忘记await：协程未被正确调度（检查所有协程调用是否都有await）

- 阻塞事件循环：在协程中使用同步I/O操作（如time.sleep()）

- 任务泄漏：创建的任务未被正确await或取消

 

2. 性能优化技巧

 

1. 批量处理：使用

"asyncio.gather"并发执行独立任务

2. 限制并发数：使用信号量控制最大并发

semaphore = asyncio.Semaphore(5) # 最大5个并发

 

async def limited_task(id):

    async with semaphore:

        print(f"任务 {id} 开始")

        await asyncio.sleep(1)

        print(f"任务 {id} 完成")

3. 合理设置超时：避免单个任务阻塞整个系统

 

六、学习路径建议

 

1. 入门阶段：

   - 掌握

"async/await"基本语法

   - 理解事件循环工作原理

   - 练习简单的并发任务

2. 进阶阶段：

   - 学习

"asyncio"高级API（Queue、Lock等）

   - 实现异步上下文管理器

   - 处理异常和任务取消

3. 生产环境：

   - 结合aiohttp、asyncpg等异步库

   - 实现完整的异步服务架构

   - 进行性能监控和优化

 

异步编程是现代Python开发的重要技能，特别适合I/O密集型应用。通过合理使用可以显著提升程序性能，但需要注意避免常见陷阱，建议从简单场景开始逐步深入。