当前位置：首页 > news >正文

Day06- (使用asyncio进行异步编程：事件循环和协程)

news 2025/7/7 10:15:00

使用`asyncio`进行异步编程：事件循环和协程

异步编程是构建高效和可扩展应用程序的关键范式，特别是在 I/O 密集型场景中。Python 的 asyncio 库提供了一个强大的框架，用于使用单个线程编写并发代码，避免了传统多线程相关的开销。本课程深入探讨 asyncio 的核心概念，重点关注事件循环和协程，并为你提供有效利用异步编程的知识。

理解异步编程

异步编程允许程序在执行多个任务时不会阻塞主线程。与同步编程（操作按顺序执行）不同，异步编程使程序能够启动一个任务，然后在等待第一个任务完成时切换到另一个任务。这种方法对于 I/O 密集型操作特别有益，例如网络请求、文件读取和数据库查询，这些操作中程序会花费大量时间等待外部资源。

同步与异步执行

为了说明区别，考虑一个需要从三个不同网站获取数据的场景。

同步: 该程序从第一个网站获取数据，等待响应，然后从第二个网站获取数据，等待响应，最后从第三个网站获取数据，等待其响应。总耗时是每个请求所耗时间的总和。
异步： 该程序同时向所有三个网站发起请求。它不会在向其他两个网站发送请求之前等待第一个网站的响应。相反，它会在响应可用时切换任务。总耗时大约是耗时最慢的请求时间，因为其他请求是并行执行的。

异步编程的优势

性能提升： 异步编程可以通过允许程序在等待 I/O 操作完成时执行其他任务，显著提高 I/O 密集型应用程序的性能。
提升可扩展性： 异步编程使应用程序能够在不阻塞主线程的情况下处理大量并发连接或请求，从而提升可扩展性。
更好的响应性： 异步编程可以通过防止 UI 或主线程被长时间运行的操作阻塞，来提高应用程序的响应性。

事件循环：`asyncio` 的核心

事件循环是 asyncio 中的核心执行机制。它负责管理和调度异步任务的执行。你可以把它想象成一个交通指挥官，它引导执行流程，确保任务高效执行且不阻塞主线程。

事件循环的工作原理

任务注册： 异步任务，以协程的形式表示，被注册到事件循环中。
任务调度： 事件循环根据任务的就绪状态调度这些任务的执行。当任务等待 I/O 操作完成或满足特定条件时，该任务被视为就绪。
任务执行： 事件循环逐个执行就绪的任务。当任务遇到 I/O 操作时，它会将控制权交还给事件循环，允许事件循环执行其他就绪的任务。
事件监控： 事件循环监控 I/O 事件，例如网络套接字变得可读或可写。当事件发生时，事件循环唤醒相应的任务并恢复其执行。
循环： 事件循环会持续迭代这些步骤，直到所有任务完成或显式停止循环。

创建和运行事件循环

import asyncioasync def main():print("Hello ...")await asyncio.sleep(1) # 模拟 I/O 操作print("... World!")# 获取当前事件循环
loop = asyncio.get_event_loop()# 运行主协程直至完成
try:loop.run_until_complete(main())
finally:loop.close() # 清理循环

解释：

asyncio.get_event_loop(): 获取当前的事件循环。如果没有事件循环在运行，它会创建一个新的。
loop.run_until_complete(main()) : 运行 main() 协程直到完成。这会启动事件循环并安排 main() 协程执行。
loop.close(): 关闭事件循环，释放其持有的任何资源。在不再需要循环时关闭它非常重要，以防止资源泄漏。

现代方法（Python 3.7+）：

从 Python 3.7 开始，运行异步代码的一种更便捷的方式是使用 asyncio.run()：

import asyncioasync def main():print("Hello ...")await asyncio.sleep(1)print("... World!")asyncio.run(main())

asyncio.run() 自动创建一个新的事件循环，运行给定的协程，并在协程完成时关闭循环。这简化了代码，并使使用 asyncio 更容易。

事件循环方法

asyncio 事件循环提供了多种管理及控制异步任务执行的方法：

run_until_complete(future): 运行事件循环，直到指定的 future（一个协程或一个任务）完成。
run_forever(): 使事件循环无限期运行，直到被明确停止。
stop(): 停止事件循环。
is_running(): 如果事件循环当前正在运行，则返回 True，否则返回 False。
create_task(coroutine): 从给定的协程创建任务，并为其调度执行。
call_later(delay, callback, *args) : 安排在指定延迟后调用回调函数。
call_soon(callback, *args): 尽快调用回调函数。

协程：异步代码的基础构建模块

协程是一种特殊的函数，可以在执行过程中被挂起和恢复。它们是 asyncio 中异步代码的基本构建模块。协程允许你编写看起来和行为都像同步代码的异步代码，从而使其更易于阅读和维护。

定义协程

协程使用 async 关键字定义：

async def my_coroutine():print("Coroutine started")await asyncio.sleep(1) # 模拟 I/O 操作print("Coroutine finished")

async 关键字表示该函数是一个协程。在协程内部，你可以使用 await 关键字来挂起协程的执行，直到一个未来（另一个协程或任务）完成。

等待协程

await 关键字用于挂起协程的执行，直到一个未来（future）完成。当你 await 一个未来时，协程将控制权交还给事件循环，允许事件循环执行其他就绪的任务。一旦未来完成，协程将从它停止的地方继续执行。

import asyncioasync def fetch_data(url):print(f"Fetching data from {url}")await asyncio.sleep(2) # 模拟网络请求print(f"Data fetched from {url}")return f"Data from {url}"async def main():data1 = await fetch_data("https://example.com/api/data1")data2 = await fetch_data("https://example.com/api/data2")print(f"Received: {data1}, {data2}")asyncio.run(main())

在这个例子中，fetch_data 是一个模拟从 URL 获取数据的协程。await asyncio.sleep(2) 这行代码使协程的执行暂停 2 秒，模拟网络请求。main 协程调用 fetch_data 两次，等待每次调用的结果。

从协程创建任务

要与其他任务并发执行协程，你需要使用 asyncio.create_task() 从协程创建一个任务：

import asyncioasync def my_coroutine(name):print(f"Coroutine {name} started")await asyncio.sleep(1)print(f"Coroutine {name} finished")async def main():task1 = asyncio.create_task(my_coroutine("Task 1"))task2 = asyncio.create_task(my_coroutine("Task 2"))await task1await task2asyncio.run(main())

在这个例子中，asyncio.create_task() 从 my_coroutine 协程创建了两个任务。await task1 和 await task2 这两行代码会等待两个任务都完成后，main 协程才会结束。这允许两个协程并发运行。

串联协程

协程可以串联起来创建复杂的异步工作流。这允许你将一个大任务分解成更小、更易于管理的协程，并按特定顺序执行它们。

import asyncioasync def prepare_data():print("Preparing data...")await asyncio.sleep(1)return "Prepared data"async def process_data(data):print(f"Processing data: {data}")await asyncio.sleep(1)return f"Processed: {data}"async def store_data(data):print(f"Storing data: {data}")await asyncio.sleep(1)print(f"Data stored: {data}")async def main():data = await prepare_data()processed_data = await process_data(data)await store_data(processed_data)asyncio.run(main())

在这个例子中，main 协程将另外三个协程 prepare_data、process_data 和 store_data 链接在一起。await 关键字确保每个协程按正确的顺序执行。

实用示例与演示

示例 1：并发网络请求

本示例演示如何使用 asyncio 进行并发网络请求。

import asyncio
import aiohttpasync def fetch_url(session, url):try:async with session.get(url) as response:return await response.text()except Exception as e:print(f"Error fetching {url}: {e}")return Noneasync def main():urls = ["https://www.example.com","https://www.python.org","https://www.google.com"]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):if result:print(f"Content from {url}: {result[:50]}...") # Print first 50 charactersasyncio.run(main())

解释：

aiohttp 是一个用于发送 HTTP 请求的异步 HTTP 客户端库。
async with aiohttp.ClientSession() as session: 创建一个异步 HTTP 会话。
async with session.get(url) as response: 向给定 URL 发送异步 GET 请求。
await response.text() 将响应体读取为文本。
asyncio.gather(*tasks) 会并发运行所有任务，并返回一个结果列表。

示例2：异步文件处理

这个例子展示了如何使用 asyncio 异步处理文件。

import asyncioasync def read_file(filename):print(f"Reading file: {filename}")await asyncio.sleep(1) # Simulate I/O delaywith open(filename, 'r') as f:content = f.read()print(f"File {filename} read")return contentasync def process_file(filename):content = await read_file(filename)print(f"Processing file: {filename}")await asyncio.sleep(1) # Simulate processing delayprocessed_content = content.upper()print(f"File {filename} processed")return processed_contentasync def write_file(filename, content):print(f"Writing to file: {filename}")await asyncio.sleep(1) # Simulate I/O delaywith open(filename, 'w') as f:f.write(content)print(f"File {filename} written")async def main():filenames = ["file1.txt", "file2.txt"]for filename in filenames:# Create dummy fileswith open(filename, 'w') as f:f.write(f"This is the content of {filename}")tasks = [process_file(filename) for filename in filenames]processed_contents = await asyncio.gather(*tasks)write_tasks = [write_file(f"processed_{filename}", content) for filename, content in zip(filenames, processed_contents)]await asyncio.gather(*write_tasks)asyncio.run(main())