汉字魔法师
118.67M · 2026-02-04
在 Python 的世界里，协程（coroutine）是实现并发和异步编程的强大工具。而 asyncio 库则是 Python 异步编程的核心。理解 asyncio 的事件循环（Event Loop）机制，是掌握 Python 异步编程的关键。本文将深入探讨 Python 协程与 asyncio 事件循环的原理、运作方式及实战应用。
协程是一种用户态的轻量级线程，它允许函数在执行过程中暂停，并在之后从暂停点继续执行。与传统线程不同，协程的切换由程序显式控制，而不是操作系统。这使得协程的开销非常小，且避免了多线程中的锁竞争问题。
Python 中通过 async def 定义协程函数，通过 await 关键字暂停协程的执行，等待一个异步操作完成。
import asyncio
async def greet(name, delay):
await asyncio.sleep(delay) # 模拟IO操作，暂停执行
print(f"Hello, {name} after {delay} seconds!")
async def main():
# 同时运行两个协程
task1 = asyncio.create_task(greet("Alice", 2))
task2 = asyncio.create_task(greet("Bob", 1))
await task1
await task2
if __name__ == "__main__":
asyncio.run(main())
asyncio 事件循环事件循环是 asyncio 的核心，它负责管理和调度所有的异步任务。可以将其想象成一个无限循环，不断地检查是否有任务准备就绪可以执行，或者是否有 I/O 操作完成。
当一个协程遇到 await 表达式时，它会暂停执行并将控制权交还给事件循环。事件循环此时不会阻塞，而是去检查其他准备就绪的协程。当 await 等待的异步操作完成后，事件循环会再次调度该协程，使其从上次暂停的地方继续执行。
这个过程通过一个称为“选择器”（Selector）的机制实现，它能高效地坚控多个 I/O 句柄的状态，例如 select、epoll 或 kqueue。
asyncio.run() 的作用asyncio.run() 是 Python 3.7+ 引入的方便函数，它负责：
import asyncio
async def my_coroutine():
print("Hello from coroutine!")
await asyncio.sleep(1)
print("Coroutine finished.")
if __name__ == "__main____":
asyncio.run(my_coroutine()) # 简化了事件循环的管理
asyncio.Task 是对协程的封装，它负责将协程注册到事件循环中，使其能够被调度和执行。通过 asyncio.create_task() 创建任务，可以将协程转化为可并发执行的对象。
import asyncio
async def say_after(delay, message):n await asyncio.sleep(delay)n print(message)nnasync def main_tasks():n task1 = asyncio.create_task(say_after(1, 'hello'))n task2 = asyncio.create_task(say_after(2, 'world'))nn print(f"started at {asyncio.get_event_loop().time()}")nn await task1n await task2nn print(f"finished at {asyncio.get_event_loop().time()}")nnif __name__ == "__main__":n asyncio.run(main_tasks())
异步编程最适合处理 I/O 密集型任务，如网络请求、数据库查询、文件读写等。在等待 I/O 完成时，CPU 可以切换到其他任务，提高效率。
在进行大量并发网络请求时，为了避免服务器过载或自身资源耗尽，可以使用 asyncio.Semaphore 来限制并发数量。
import asyncio
import aiohttp
async def fetch_url(session, url, semaphore):
async with semaphore:
async with session.get(url) as response:
return await response.text()
async def main_semaphore(urls, limit=5):
semaphore = asyncio.Semaphore(limit)
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url, semaphore) for url in urls]
results = await asyncio.gather(*tasks)
return results
if __name__ == "__main__":
test_urls = ["http://example.com/" for _ in range(10)]
asyncio.run(main_semaphore(test_urls))
使用 try...except 块来处理协程中的异常，确保一个协程的失败不会导致整个事件循环崩溃。
asyncio 提供了强大的调试功能，可以通过设置 ASYNCIO_DEBUG=1 环境变量或 loop.set_debug(True) 来开启。
Python 的 asyncio 及其事件循环机制为开发者提供了构建高性能、高并发应用的利器。通过理解协程、任务和事件循环的内在联系，并遵循最佳实践，我们能够编写出更健壮、更高效的异步 Python 代码，从而更好地应对现代 Web 开发中的挑战。
希望本文能帮助你更好地掌握 Python 协程和 asyncio 事件循环的奥秘！
