Python asyncio 用法深度解析

2025-09-02 15:12:53  阅读 17 次 评论 0 条

引言

在现代软件开发中，处理高并发I/O操作是一个常见挑战。传统的同步编程模型在面对大量网络请求、文件操作或数据库查询时往往效率低下。Python的asyncio库应运而生，它提供了基于async/await语法的异步编程能力，让开发者能够编写高效、可扩展的并发代码。

本文将深入探讨asyncio的核心概念、用法、最佳实践以及常见陷阱，帮助你全面掌握这一强大的并发编程工具。

一、核心概念解析

1.1 事件循环 (Event Loop)

事件循环是asyncio的心脏，它负责调度和执行异步任务。想象事件循环就像一个高效的交通指挥员，它不断检查哪些协程可以运行，并在它们之间智能切换。

import asyncio

async def main():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

# 现代Python推荐方式（3.7+）
asyncio.run(main())

# 传统方式（了解即可）
loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main())
finally:
    loop.close()

1.2 协程 (Coroutine)

协程是asyncio的基本执行单元，使用async def定义。它们与普通函数的区别在于能够被”暂停”和”恢复”。

async def fetch_data():
    print("开始获取数据")
    await asyncio.sleep(2)  # 模拟I/O操作
    print("数据获取完成")
    return {"data": "sample"}

# 调用协程必须使用await
async def main():
    result = await fetch_data()
    print(f"结果: {result}")

1.3 Future 和 Task

  • Future: 表示异步操作的最终结果，类似于JavaScript中的Promise
  • Task: Future的子类，专门用于包装和管理协程
async def my_task():
    await asyncio.sleep(1)
    return "任务完成"

async def main():
    # 创建任务
    task = asyncio.create_task(my_task())
    
    # 可以继续执行其他操作
    print("任务已创建，继续其他工作...")
    
    # 等待任务完成
    result = await task
    print(f"任务结果: {result}")

二、基本用法详解

2.1 并发执行多个任务

asyncio的真正威力在于能够轻松实现并发操作：

import asyncio
import time

async def download_file(filename, duration):
    print(f"开始下载 {filename}")
    await asyncio.sleep(duration)
    print(f"{filename} 下载完成")
    return f"{filename}_content"

async def main():
    start_time = time.time()
    
    # 方法1: 使用gather（推荐）
    results = await asyncio.gather(
        download_file("file1.txt", 2),
        download_file("file2.txt", 3),
        download_file("file3.txt", 1)
    )
    print(f"所有文件下载完成: {results}")
    
    # 方法2: 分别创建任务
    task1 = asyncio.create_task(download_file("file4.txt", 2))
    task2 = asyncio.create_task(download_file("file5.txt", 1))
    
    await task1
    await task2
    
    end_time = time.time()
    print(f"总耗时: {end_time - start_time:.2f}秒")

asyncio.run(main())

2.2 超时处理

在实际应用中，超时控制至关重要：

async def slow_operation():
    await asyncio.sleep(10)
    return "操作完成"

async def main():
    try:
        # 设置5秒超时
        result = await asyncio.wait_for(slow_operation(), timeout=5.0)
        print(result)
    except asyncio.TimeoutError:
        print("操作超时，已取消")
    
    # Python 3.11+ 的新语法
    try:
        async with asyncio.timeout(5.0):
            result = await slow_operation()
            print(result)
    except TimeoutError:
        print("操作超时")

asyncio.run(main())

2.3 同步原语

asyncio提供了线程类似的同步机制：

async def worker(lock, name, work_time):
    async with lock:  # 获取锁
        print(f"{name} 开始工作")
        await asyncio.sleep(work_time)
        print(f"{name} 工作完成")

async def main():
    lock = asyncio.Lock()
    
    # 三个 worker 竞争同一个锁
    await asyncio.gather(
        worker(lock, "Worker1", 2),
        worker(lock, "Worker2", 1),
        worker(lock, "Worker3", 3)
    )

asyncio.run(main())

三、高级用法

3.1 队列 (Queue)

生产者-消费者模式是并发编程的常见模式：

async def producer(queue, items):
    for item in items:
        await queue.put(item)
        print(f"生产了: {item}")
        await asyncio.sleep(0.1)  # 模拟生产时间
    await queue.put(None)  # 结束信号

async def consumer(queue, name):
    while True:
        item = await queue.get()
        if item is None:
            queue.put(None)  # 传递给其他消费者
            print(f"{name} 结束工作")
            break
        
        print(f"{name} 消费了: {item}")
        await asyncio.sleep(0.2)  # 模拟处理时间
        queue.task_done()  # 标记任务完成

async def main():
    queue = asyncio.Queue(maxsize=3)  # 限制队列大小
    
    # 启动生产者和消费者
    producer_task = asyncio.create_task(producer(queue, range(5)))
    consumer_tasks = [
        asyncio.create_task(consumer(queue, "消费者1")),
        asyncio.create_task(consumer(queue, "消费者2"))
    ]
    
    await producer_task
    await queue.join()  # 等待所有任务处理完成
    
    for task in consumer_tasks:
        task.cancel()
    
    # 等待消费者任务正常结束
    await asyncio.gather(*consumer_tasks, return_exceptions=True)

asyncio.run(main())

3.2 事件 (Event) 和条件 (Condition)

async def waiter(event, name):
    print(f"{name} 等待事件触发")
    await event.wait()
    print(f"{name} 检测到事件，继续执行")

async def setter(event):
    await asyncio.sleep(2)
    print("设置事件")
    event.set()  # 触发所有等待的协程

async def main():
    event = asyncio.Event()
    
    await asyncio.gather(
        waiter(event, "任务1"),
        waiter(event, "任务2"),
        setter(event)
    )

asyncio.run(main())

3.3 子进程管理

async def run_command():
    # 创建子进程
    process = await asyncio.create_subprocess_exec(
        'python', '--version',
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    
    # 等待进程完成并获取输出
    stdout, stderr = await process.communicate()
    
    print(f"退出码: {process.returncode}")
    if stdout:
        print(f"标准输出: {stdout.decode()}")
    if stderr:
        print(f"错误输出: {stderr.decode()}")

asyncio.run(run_command())

四、实际应用示例

4.1 并发HTTP请求

import aiohttp
import asyncio

async def fetch_url(session, url):
    try:
        async with session.get(url, timeout=10) as response:
            return await response.text()
    except Exception as e:
        return f"错误: {e}"

async def main():
    urls = [
        "http://httpbin.org/get",
        "http://httpbin.org/delay/2",
        "http://httpbin.org/status/404"
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        
        for url, result in zip(urls, results):
            print(f"{url}: {result[:100]}...")

asyncio.run(main())

4.2 WebSocket客户端

import aiohttp
import asyncio

async def websocket_client():
    try:
        async with aiohttp.ClientSession() as session:
            async with session.ws_connect("ws://echo.websocket.org") as ws:
                print("连接已建立")
                
                # 发送消息
                await ws.send_str("Hello, WebSocket!")
                
                # 接收消息
                async for msg in ws:
                    if msg.type == aiohttp.WSMsgType.TEXT:
                        print(f"收到消息: {msg.data}")
                        if msg.data == "close":
                            await ws.close()
                            break
                    elif msg.type == aiohttp.WSMsgType.CLOSED:
                        print("连接关闭")
                        break
                    elif msg.type == aiohttp.WSMsgType.ERROR:
                        print("连接错误")
                        break
    except Exception as e:
        print(f"连接失败: {e}")

asyncio.run(websocket_client())

五、注意事项与最佳实践

5.1 避免阻塞事件循环

这是asyncio开发中最常见的错误：

# ❌ 错误示例 - 阻塞调用
async def bad_example():
    import time
    time.sleep(5)  # 这会阻塞整个事件循环！

# ✅ 正确做法 - 使用异步版本
async def good_example():
    await asyncio.sleep(5)  # 非阻塞

# ✅ 对于无法异步的阻塞操作
async def run_blocking():
    loop = asyncio.get_running_loop()
    # 在线程池中运行阻塞函数
    result = await loop.run_in_executor(None, time.sleep, 5)
    return result

5.2 正确的错误处理

async def risky_operation():
    try:
        await asyncio.sleep(1)
        raise ValueError("模拟错误")
    except ValueError as e:
        print(f"操作失败: {e}")
        raise  # 重新抛出

async def main():
    # 单个任务的错误处理
    try:
        await risky_operation()
    except ValueError as e:
        print(f"捕获错误: {e}")
    
    # 多个任务的错误处理
    tasks = [
        asyncio.create_task(risky_operation()),
        asyncio.create_task(asyncio.sleep(2))
    ]
    
    done, pending = await asyncio.wait(tasks, return_exceptions=True)
    
    for task in done:
        if isinstance(task, Exception):
            print(f"任务失败: {task}")
        else:
            print(f"任务成功: {task.result()}")

asyncio.run(main())

5.3 资源清理

async def fetch_with_resource():
    # 模拟资源获取
    resource = "connection"
    print(f"获取资源: {resource}")
    
    try:
        # 模拟业务操作
        await asyncio.sleep(1)
        return "success"
    finally:
        # 确保资源释放
        print(f"释放资源: {resource}")

async def main():
    result = await fetch_with_resource()
    print(f"操作结果: {result}")

asyncio.run(main())

5.4 调试技巧

async def debug_example():
    # 获取当前任务信息
    task = asyncio.current_task()
    print(f"任务名称: {task.get_name()}")
    print(f"协程: {task.get_coro()}")
    
    await asyncio.sleep(1)

async def main():
    # 启用调试模式
    asyncio.run(debug_example(), debug=True)
    
    # 或者设置事件循环的调试模式
    loop = asyncio.get_event_loop()
    loop.set_debug(True)
    
    try:
        await debug_example()
    finally:
        loop.set_debug(False)

asyncio.run(main())

六、性能优化建议

6.1 限制并发数

async def limited_concurrency(urls, max_concurrent=5):
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def fetch_with_limit(url):
        async with semaphore:
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as response:
                    return await response.text()
    
    tasks = [fetch_with_limit(url) for url in urls]
    return await asyncio.gather(*tasks)

# 使用示例
async def main():
    urls = ["http://example.com"] * 100
    results = await limited_concurrency(urls, max_concurrent=10)
    print(f"完成了 {len(results)} 个请求")

asyncio.run(main())

6.2 使用连接池

async def efficient_requests():
    connector = aiohttp.TCPConnector(limit=10)  # 限制连接数
    
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [session.get(f"http://example.com/page/{i}") for i in range(100)]
        responses = await asyncio.gather(*tasks)
        
        # 处理响应
        results = []
        for response in responses:
            results.append(await response.text())
        
        return results

6.3 批量操作

async def batch_operations():
    # 而不是这样：
    # results = []
    # for item in items:
    #     result = await process_item(item)
    #     results.append(result)
    
    # 应该这样：
    tasks = [process_item(item) for item in items]
    results = await asyncio.gather(*tasks)
    return results

七、常见问题解答

Q1: 如何从同步代码调用异步函数？

def sync_wrapper():
    # 方法1: 使用asyncio.run（适用于简单脚本）
    result = asyncio.run(async_function())
    return result

# 方法2: 在已有事件循环中运行
def sync_in_loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        result = loop.run_until_complete(async_function())
        return result
    finally:
        loop.close()

Q2: 如何正确取消任务？

async def cancellable_task():
    try:
        await asyncio.sleep(10)
        print("任务正常完成")
        return "success"
    except asyncio.CancelledError:
        print("任务被取消，执行清理操作")
        raise  # 重新抛出异常

async def main():
    task = asyncio.create_task(cancellable_task())
    
    # 等待一段时间后取消
    await asyncio.sleep(2)
    task.cancel()
    
    try:
        await task
    except asyncio.CancelledError:
        print("任务已成功取消")

asyncio.run(main())

Q3: 如何处理任务中的异常？

async def failing_task():
    await asyncio.sleep(1)
    raise ValueError("任务失败")

async def main():
    task = asyncio.create_task(failing_task())
    
    try:
        await task
    except ValueError as e:
        print(f"捕获到异常: {e}")
        print(f"任务异常: {task.exception()}")

asyncio.run(main())

关键点

  1. 理解事件循环机制 - 这是异步编程的基础
  2. 正确使用async/await - 避免阻塞调用
  3. 合理管理并发 - 使用信号量限制并发数
  4. 妥善处理错误 - 确保程序的稳定性
  5. 注意资源清理 - 防止资源泄漏
本文地址：http://bloguan.com/?id=628
版权声明：本文为原创文章，版权归 博观网 所有，欢迎分享本文，转载请保留出处！
NEXT:已经是最新一篇了
相关文章

发表评论


表情

还没有留言，还不快点抢沙发？