当前位置：首页 > 文章列表 > 文章 > python教程 > asyncio异步编程基础教程

asyncio异步编程基础教程

2025-09-18 13:04:05 0浏览收藏

小伙伴们对文章编程感兴趣吗？是否正在学习相关知识点？如果是，那么本文《asyncio异步编程入门教程》，就很适合你，本篇文章讲解的知识点主要包括。在之后的文章中也会多多分享相关知识点，希望对大家的知识积累有所帮助！

asyncio通过协程实现单线程并发，适用于I/O密集型任务。使用async/await定义和调用协程，通过事件循环调度执行。可用asyncio.run()启动主协程，create_task()并发运行多个协程，gather()等待所有协程完成。异常处理需在await时捕获，未处理异常会存储于Task中。避免阻塞事件循环：使用异步I/O、将CPU密集型任务放入线程或进程池、用wait_for()设置超时、定期调用sleep(0)让出控制权。相比线程和进程，asyncio轻量高效，适合I/O密集场景；CPU密集任务应选进程或多线程；混合任务可结合使用三种方式。

如何使用asyncio进行异步编程？

Asyncio允许你编写并发代码，而无需线程或进程的复杂性。它本质上是一个单线程、事件循环的并发框架，通过协程（coroutines）实现非阻塞操作，从而提升I/O密集型任务的效率。

使用asyncio进行异步编程，简单来说，就是用async和await关键字定义和调用协程，并将它们注册到事件循环中执行。

解决方案：

定义协程： 使用async def定义一个协程函数。协程可以暂停执行，等待I/O操作完成，而不会阻塞整个程序。

import asyncio

async def my_coroutine(delay):
    print(f"协程开始，等待 {delay} 秒")
    await asyncio.sleep(delay)
    print("协程结束")
    return f"等待了 {delay} 秒的结果"

创建事件循环： asyncio的核心是事件循环，它负责调度和执行协程。
```
loop = asyncio.get_event_loop()
```
运行协程： 可以使用loop.run_until_complete()或asyncio.run()来运行协程。asyncio.run()简化了事件循环的创建和关闭，更适合简单的脚本。
```
async def main():
    result = await my_coroutine(2)
    print(result)

asyncio.run(main())
```

使用await关键字： await关键字用于等待一个awaitable对象（例如另一个协程、Future或Task）完成。await只能在async函数中使用。

async def another_coroutine():
    await asyncio.sleep(1)
    return "另一个协程的结果"

async def main():
    result = await another_coroutine()
    print(result)

asyncio.run(main())

创建Tasks： asyncio.create_task()函数可以创建一个Task对象，它代表一个正在运行的协程。这允许你并发地运行多个协程。

async def main():
    task1 = asyncio.create_task(my_coroutine(1))
    task2 = asyncio.create_task(my_coroutine(2))

    await task1
    await task2

    print("所有任务完成")

asyncio.run(main())

并发执行多个协程： 使用asyncio.gather()可以并发地运行多个协程，并等待它们全部完成。

async def main():
    results = await asyncio.gather(
        my_coroutine(1),
        my_coroutine(2),
        another_coroutine()
    )
    print(results)

asyncio.run(main())

如何处理asyncio中的异常？

在asyncio中处理异常与在常规Python代码中类似，但需要注意协程的特性。可以使用try...except块来捕获协程中抛出的异常。

import asyncio

async def might_fail():
    await asyncio.sleep(0.5)
    raise ValueError("Something went wrong!")

async def main():
    try:
        await might_fail()
    except ValueError as e:
        print(f"Caught an error: {e}")

asyncio.run(main())

如果一个任务（Task）内部抛出了未捕获的异常，该异常会被存储在Task对象中。当使用await等待该Task完成时，异常会被重新抛出。如果Task永远没有被await，则异常会被事件循环记录下来，但不会中断程序。

import asyncio

async def task_that_fails():
    await asyncio.sleep(0.5)
    raise ValueError("Task failed!")

async def main():
    task = asyncio.create_task(task_that_fails())
    await asyncio.sleep(1) # 让任务有时间完成
    try:
        await task # 重新抛出异常
    except ValueError as e:
        print(f"Caught error from task: {e}")

asyncio.run(main())

可以使用task.result()方法来获取Task的结果或异常。如果Task已经完成并且没有抛出异常，则返回结果。如果Task抛出了异常，则该方法会重新抛出异常。

import asyncio

async def task_that_fails():
    await asyncio.sleep(0.5)
    raise ValueError("Task failed!")

async def main():
    task = asyncio.create_task(task_that_fails())
    await asyncio.sleep(1)
    try:
        task.result() # 重新抛出异常
    except ValueError as e:
        print(f"Caught error from task: {e}")

asyncio.run(main())

asyncio中如何避免阻塞事件循环？

避免阻塞asyncio事件循环是保证程序响应性的关键。事件循环是单线程的，如果一个协程执行了耗时的同步操作，它会阻塞整个事件循环，导致其他协程无法运行。

使用异步I/O操作： 这是最基本也是最重要的原则。使用asyncio提供的异步I/O操作，例如asyncio.sleep()、asyncio.open_connection()、aiohttp（异步HTTP客户端）等。这些操作会在等待I/O完成时释放事件循环的控制权，允许其他协程运行。

避免CPU密集型任务： 如果需要执行CPU密集型任务（例如图像处理、加密解密等），应该将这些任务放到单独的进程或线程中执行，避免阻塞事件循环。可以使用asyncio.to_thread()或concurrent.futures模块来实现。

import asyncio
import concurrent.futures
import time

def cpu_bound_task(n):
    time.sleep(n)  # 模拟耗时操作
    return f"CPU密集型任务完成，耗时 {n} 秒"

async def main():
    loop = asyncio.get_running_loop()
    with concurrent.futures.ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_bound_task, 2) # 放在进程池中运行
        print(result)

asyncio.run(main())

限制协程的运行时间： 使用asyncio.wait_for()可以限制一个协程的运行时间。如果协程在指定时间内没有完成，asyncio.wait_for()会抛出一个asyncio.TimeoutError异常，可以捕获并处理该异常。

import asyncio

async def long_running_task():
    await asyncio.sleep(5)
    return "任务完成"

async def main():
    try:
        result = await asyncio.wait_for(long_running_task(), timeout=2)
        print(result)
    except asyncio.TimeoutError:
        print("任务超时")

asyncio.run(main())

使用异步库： 尽量使用异步库来替代同步库。例如，使用aiohttp替代requests，使用asyncpg或aiosqlite替代psycopg2或sqlite3。

定期释放控制权： 如果一个协程需要执行大量的计算或I/O操作，可以定期使用await asyncio.sleep(0)来释放事件循环的控制权，让其他协程有机会运行。

import asyncio

async def large_task():
    for i in range(100000):
        # 模拟大量计算
        _ = i * i
        if i % 1000 == 0:
            await asyncio.sleep(0)  # 释放控制权
    return "任务完成"

async def main():
    result = await large_task()
    print(result)

asyncio.run(main())

如何选择合适的并发策略：asyncio vs 线程 vs 进程？

选择合适的并发策略取决于任务的性质和程序的需求。Asyncio、线程和进程各有优缺点，适用于不同的场景。

Asyncio：
- 优点：
  - 轻量级： 基于协程，切换开销小，资源占用少。
  - 高效I/O： 擅长处理I/O密集型任务，例如网络请求、文件读写等。
  - 单线程： 避免了线程锁的复杂性，减少了死锁和竞争条件。
- 缺点：
  - CPU密集型： 不适合CPU密集型任务，因为单线程无法充分利用多核CPU。
  - 阻塞操作： 如果有阻塞操作，会阻塞整个事件循环，影响程序的响应性。
- 适用场景：
  - Web服务器、网络爬虫、聊天服务器、实时数据处理等I/O密集型应用。
线程：
- 优点：
  - 并发执行： 可以并发执行CPU密集型任务，充分利用多核CPU。
  - 共享内存： 线程之间可以共享内存，方便数据交换。
- 缺点：
  - 线程锁： 需要使用线程锁来保护共享资源，增加了编程复杂性。
  - GIL限制： 在CPython中，由于全局解释器锁（GIL）的存在，同一时刻只能有一个线程执行Python字节码，限制了CPU密集型任务的并发性能。
  - 上下文切换： 线程切换开销比协程大。
- 适用场景：
  - GUI程序、多线程下载器、需要并发执行CPU密集型任务的应用。
进程：
- 优点：
  - 真正的并行： 可以利用多核CPU实现真正的并行，不受GIL限制。
  - 隔离性： 进程之间相互隔离，一个进程崩溃不会影响其他进程。
- 缺点：
  - 资源占用： 进程创建和销毁开销大，资源占用多。
  - 通信开销： 进程间通信（IPC）开销比线程大。
- 适用场景：
  - 科学计算、图像处理、视频编码等CPU密集型应用，需要高可靠性和隔离性的应用。