Python Asyncio 教程：理解事件循环、任务调度与非阻塞暂停-Python教程-PHP中文网

Python Asyncio 教程：理解事件循环、任务调度与非阻塞暂停

花韻仙語

发布： 2025-12-02 11:48:02

原创

808人浏览过

Python Asyncio 教程：理解事件循环、任务调度与非阻塞暂停

本文深入探讨python `asyncio`异步编程中一个常见误区：在异步代码中使用`time.sleep`导致事件循环阻塞。我们将阐明`asyncio`的单线程协作式并发机制，解释为何必须通过`await`关键字显式让出控制权。教程将详细介绍如何利用`await asyncio.sleep()`实现非阻塞暂停，并提供正确的`asyncio`程序结构与事件循环管理实践，确保并发任务按预期运行。

1. Asyncio 核心概念：协作式并发

Python的asyncio库是实现单线程并发编程的强大工具，尤其适用于I/O密集型任务。与多线程编程通过操作系统调度实现并行执行不同，asyncio采用的是协作式多任务处理模型。这意味着在asyncio应用程序中，所有协程（coroutine）都在同一个线程中运行，它们必须通过显式地让出控制权（通常通过await关键字）来允许事件循环调度其他协程执行。

事件循环是asyncio的核心，它负责监听各种事件（如网络I/O完成、定时器到期等），并调度相应的协程执行。当一个协程遇到一个耗时操作（如网络请求、文件读写）时，如果这个操作是异步的，它会使用await关键字暂停自身的执行，并将控制权交还给事件循环。事件循环此时可以去执行其他已准备好的协程，待原协程等待的I/O操作完成后，事件循环再将其唤醒继续执行。这种机制使得单个线程能够高效地处理大量并发任务，而无需创建和管理多个线程。

2. time.sleep 的陷阱：阻塞事件循环

在asyncio上下文中，一个常见的错误是使用Python标准库中的time.sleep()函数来引入延迟。time.sleep()是一个同步阻塞函数，它的作用是暂停当前正在执行的整个线程，直到指定的秒数过去。

当asyncio事件循环正在运行时，如果在任何协程或主程序中调用了time.sleep()，它将立即阻塞整个线程。这意味着事件循环将无法继续调度其他协程，也无法处理任何I/O事件。所有通过asyncio.create_task()创建并计划执行的协程都将停滞不前，直到time.sleep()完成。这与asyncio的协作式并发精神背道而驰，导致应用程序失去响应性，无法实现真正的并发。

立即学习“Python免费学习笔记（深入）”；

例如，如果在一个异步主函数中，创建了一个后台任务，然后紧接着使用time.sleep()进行等待，那么后台任务将无法获得执行机会，因为主线程被time.sleep()完全阻塞了。

3. 正确的非阻塞暂停：await asyncio.sleep()

为了在asyncio应用程序中实现非阻塞的暂停，我们必须使用asyncio库提供的异步版本：asyncio.sleep()。

asyncio.sleep()是一个协程函数，它接受一个秒数作为参数，并返回一个可等待对象。当一个协程执行到await asyncio.sleep(delay)时，它会暂停自身的执行，并将控制权交还给事件循环。事件循环会将这个协程标记为“等待delay秒后唤醒”，然后立即去调度执行其他已准备好的协程。当delay时间过去后，事件循环会重新唤醒该协程，使其从暂停的地方继续执行。

这种机制确保了在等待期间，事件循环能够保持活跃，持续处理其他并发任务，从而维持应用程序的响应性和并发性。

4. 示例：构建一个响应式的 Asyncio 应用

下面是一个完整的asyncio应用程序示例，演示了如何正确地创建任务、管理事件循环以及使用await asyncio.sleep()实现非阻塞暂停。

import asyncio
import time

# 这是一个异步协程，模拟一个耗时操作
async def background_task(task_id: int):
    """一个模拟后台工作的协程"""
    print(f"任务 {task_id}: 启动...")
    await asyncio.sleep(task_id * 0.5)  # 使用 asyncio.sleep 进行非阻塞等待
    print(f"任务 {task_id}: 完成!")

# 异步主函数，作为程序的入口点
async def main():
    """主函数，负责创建和管理异步任务"""
    print("主程序: 启动")

    # 创建多个后台任务，它们将并发运行
    task1 = asyncio.create_task(background_task(1))
    task2 = asyncio.create_task(background_task(2))
    task3 = asyncio.create_task(background_task(3))

    print("主程序: 后台任务已调度，现在等待它们完成...")

    # 在主循环中进行非阻塞等待，允许后台任务运行
    # 如果这里使用 time.sleep(0.1)，所有后台任务都将无法执行
    for i in range(1, 7):
        print(f"主程序: 正在执行第 {i} 轮主循环...")
        await asyncio.sleep(0.5) # 非阻塞等待，将控制权交还给事件循环
        # 可以在这里检查任务状态，例如：
        # if task1.done():
        #     print(f"任务 1 已完成，结果: {task1.result()}")

    print("主程序: 等待所有任务最终完成...")
    # 确保所有后台任务都已完成
    await asyncio.gather(task1, task2, task3)

    print("主程序: 所有任务已完成，程序退出。")

# 使用 asyncio.run() 启动事件循环并运行主函数
if __name__ == "__main__":
    start_time = time.monotonic()
    asyncio.run(main())
    end_time = time.monotonic()
    print(f"总运行时间: {end_time - start_time:.2f} 秒")

登录后复制

代码说明：

腾讯Effidit

腾讯AI Lab开发的AI写作助手，提升写作者的写作效率和创作体验

查看详情

background_task：这是一个简单的异步协程，它使用await asyncio.sleep()来模拟一个耗时操作。
main：这是一个异步主函数，它是asyncio应用程序的推荐入口点。
- asyncio.create_task()：用于将协程包装成一个Task对象，并将其调度到事件循环中运行。任务一旦创建，事件循环就会在适当的时机开始执行它。
- await asyncio.sleep(0.5)：在主循环中，我们使用await asyncio.sleep()来周期性地让出控制权。这使得事件循环有机会去执行background_task以及其他任何已调度但尚未完成的任务。
- await asyncio.gather(task1, task2, task3)：这是一个非常有用的函数，它会等待所有传入的协程或任务完成。在这里，我们用它来确保在main函数退出之前，所有的后台任务都已执行完毕。
asyncio.run(main())：这是启动asyncio事件循环并运行主异步函数的最简单和推荐方式。它负责创建事件循环、运行main()协程，并在main()完成后关闭事件循环。

通过运行上述代码，您会观察到background_task的输出与main函数的循环输出交织在一起，这证明了所有任务都在并发执行，并且main函数没有被阻塞。

5. 避免同步阻塞：注意事项

为了充分利用asyncio的优势，以下是一些重要的注意事项：

始终使用await关键字：在异步函数中，任何可能导致I/O阻塞或长时间等待的操作（如网络请求、文件I/O、数据库查询等），都必须使用await关键字来调用其异步版本。如果一个库只提供同步API，那么直接调用它将阻塞事件循环。
避免在异步函数中直接调用同步阻塞函数：如time.sleep()、requests.get()（同步版本）、open().read()（同步文件I/O）等。

处理同步阻塞任务：如果您的应用程序确实需要执行一些CPU密集型或只能通过同步API访问的阻塞操作，可以考虑使用loop.run_in_executor。这会将阻塞操作提交到一个单独的线程池或进程池中执行，从而避免阻塞主事件循环。

import concurrent.futures

async def run_blocking_code():
    # 假设 some_blocking_function 是一个同步的、耗时函数
    result = await asyncio.get_event_loop().run_in_executor(
        concurrent.futures.ThreadPoolExecutor(), # 或 ProcessPoolExecutor()
        some_blocking_function,
        arg1, arg2
    )
    return result

登录后复制