Python 最强异步编程：Asyncio

数据STUDIO

发布于 2024-07-24 09:33:04

1640

发布于 2024-07-24 09:33:04

文章被收录于专栏：数据STUDIO

着现代网络应用的日益复杂，处理大量并发I/O操作成为了一个挑战。Python标准库中的asyncio模块就是解决这个问题的利器。它提供了编写简洁、高效和可扩展异步代码的框架，特别适用于网络编程场景。虽然asyncio的事件循环、协程、Future等概念初听起来有些陌生，但一旦掌握了它的窍门，就会体会到它的强大之处。让我们通过实例逐步展开探索。

Asyncio基本概念

事件循环(Event Loop): asyncio的核心，负责管理和调度不同任务的执行、处理事件以及分配资源。
协程(Coroutine): 使用async/await语法定义的函数，可以在特定点暂停和恢复执行，从而允许其他操作在暂停期间运行。
Future: 代表未来结果的对象，通常由低层异步回调产生。
Task: 将协程包装为Future对象的异步执行单元，由事件循环进行调度。
更多优质内容，请关注@公众号：数据STUDIO

Asyncio 入门

Asyncio异步编程的核心思想是让程序在等待I/O操作完成的同时，可以继续执行其他任务，从而提高资源利用率。这就好比一个厨师在炖菜的同时，开始准备沙拉，而不是煮一道菜时傻站着等待。通过合理安排，程序可以在单线程下高效完成诸多任务，从而达到"伪并行"的效果，提高了性能。

await关键字

Python 3.5 中引入了异步编程，await 是其中的关键字之一。它能够暂停一个 async 函数的执行，直到可等待对象(如协程、任务、期货或I/O操作)完成，从而让出执行权，使其他任务得以在此期间运行。这一特性使得异步编程在处理I/O密集型任务和高级网络代码结构时能够高效运行。

await 只能在 async 函数内使用，否则会导致语法错误。
它的主要目的是将控制权交还给事件循环，暂停所在的协程执行，直到被等待的对象就绪。这种非阻塞方式使得异步编程高效，尤其适用于I/O密集型任务。
可与 await 一起使用的对象必须是"可等待的"。最常见的是使用 async def 声明的协程，但也包括 asyncio 的任务、期货，以及任何实现了 await() 方法的对象。

Asyncio应用

"Hello, Async World!"

想象一下，你的任务是在停顿 2 秒后打印 "Hello, World!"。异步方法很简单：

import time

def say_hello():
    time.sleep(2)
    print("Hello, Async World? (not yet)")

say_hello()

它完成了工作，但在等待这 2 秒的过程中，一切都停止了。

现在，切换到 asyncio，展示异步方式：

import asyncio

async def say_hello_async():
    await asyncio.sleep(2)
    print("Hello, Async World!")

asyncio.run(say_hello_async())

有了 asyncio，当我们等待时，事件循环可以执行其他任务，如检查电子邮件或播放音乐，从而使我们的代码不阻塞，效率更高：

import asyncio

async def say_hello_async():
    await asyncio.sleep(2)  # Simulates waiting for 2 seconds
    print("Hello, Async World!")

async def do_something_else():
    print("Starting another task...")
    await asyncio.sleep(1)  # Simulates doing something else for 1 second
    print("Finished another task!")

async def main():
    # Schedule both tasks to run concurrently
    await asyncio.gather(
        say_hello_async(),
        do_something_else(),
    )

asyncio.run(main())

在此修改版本中，main() 函数使用 asyncio.gather() 并发运行 say_hello_async() 和 do_something_else()。这意味着程序在等待 say_hello_async() 函数完成 2 秒钟的休眠时，会启动并可能完成 do_something_else() 函数，从而在等待时间内有效地执行另一项任务。

抓取网页（并发 I/O 任务）

抓取网页是展示异步编程能力的一个经典例子。让我们比较一下同步和异步获取 URL 的方式。

同步 HTTP 请求主要由 requests 库完成，连续获取两个网页的过程如下所示：

import requests
import time

start_time = time.time()

def fetch(url):
    return requests.get(url).text

page1 = fetch('http://example.com')
page2 = fetch('http://example.org')

print(f"Done in {time.time() - start_time} seconds")

# Output: Done in 0.6225857734680176 seconds

这段代码简单得不能再简单，但它会在每个请求完成后才开始下一个请求。

我们用 aiohttp 和 asyncio 来提高效率，它们可用于异步 HTTP 请求：

import aiohttp
import asyncio
import time

async def fetch_async(url, session):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        page1 = asyncio.create_task(fetch_async('http://example.com', session))
        page2 = asyncio.create_task(fetch_async('http://example.org', session))
        await asyncio.gather(page1, page2)

start_time = time.time()
asyncio.run(main())
print(f"Done in {time.time() - start_time} seconds")

# Output: Done in 0.2990539073944092 seconds

该异步版本无需等待。在获取一个页面的同时，它会开始获取下一个页面，从而大大缩短了总等待时间。

并发读取文件（I/O 任务）

我们从网络请求出发，探索了使用 asyncio 并发执行的不同用例。现在，让我们专注于异步读取多个文件。这在处理大文件或纯I/O密集型任务时尤为有用。

在同步环境下，逐个读取多个文件会大大增加执行时间，尤其是处理大文件时。示例如下:

# 同步读取多个文件
def read_file_sync(filepath):
    with open(filepath, 'r') as file:
        return file.read()

def read_all_sync(filepaths):
    return [read_file_sync(filepath) for filepath in filepaths]

filepaths = ['file1.txt', 'file2.txt']
data = read_all_sync(filepaths)
print(data)

使用异步方式，我们可以同时读取多个文件，而不会阻塞事件循环。通过让出控制权并在I/O操作完成时重新调度，异步方法可以显著提高多文件读取的效率。

对于异步版本，我们将使用 aiofiles，这是一个支持异步文件操作的库。如果尚未安装 aiofiles，可以使用 pip 安装：

pip install aiofiles

使用 aiofiles 后，我们可以在不阻塞事件循环的情况下执行文件 I/O 操作，从而可以同时读取多个文件。

import asyncio
import aiofiles

# 异步读取单个文件
async def read_file_async(filepath):
    async with aiofiles.open(filepath, 'r') as file:
        return await file.read()

async def read_all_async(filepaths):
    tasks = [read_file_async(filepath) for filepath in filepaths]
    return await asyncio.gather(*tasks)

# 运行异步函数
async def main():
    filepaths = ['file1.txt', 'file2.txt']
    data = await read_all_async(filepaths)
    print(data)

asyncio.run(main())

异步版本利用 aiofiles 和 asyncio.gather 允许并发读取多个文件。与逐个读取文件的同步版本相比，这种方法大大缩短了总执行时间。通过并发执行 I/O 操作，我们可以提高需要处理多个文件操作的程序的效率。

混合同步与同步：混合方法

有时，你无法摆脱同步函数，但仍想享受异步的乐趣。下面就是混合使用它们的方法：

import asyncio
import time

def sync_task():
    print("Starting a slow sync task...")
    time.sleep(5)  # 模拟长时间任务
    print("Finished the slow task.")

async def async_wrapper():
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, sync_task)

async def main():
    await asyncio.gather(
        async_wrapper(),
        # 想象一下其他异步任务
    )

asyncio.run(main())

所提供的代码片段演示了如何使用 Python 的 asyncio 库在异步环境中集成同步函数。更多优质内容，请关注@公众号：数据STUDIO

解释代码：

1. 异步封装器 (async_wrapper 函数):

这个异步函数演示了如何在不阻塞事件循环的情况下，以非阻塞的方式运行同步的 sync_task。它利用了loop.run_in_executor(None, sync_task)来实现这一点。
loop.run_in_executor(None, sync_task)会根据所使用的执行器，将sync_task安排在一个单独的线程或进程中运行。当第一个参数为None时，默认使用线程池执行器来运行任务。
await关键字用于等待sync_task完成执行，而不会阻塞事件循环，从而允许其他异步操作在此期间继续进行。

2. 异步执行（ main 函数）：

main函数是一个异步函数，展示了如何同时运行同步和异步任务，而不会产生阻塞。
asyncio.gather用于安排async_wrapper和其他潜在的异步任务的并发执行。通过使用gather，可以确保事件循环能够有效管理多个任务，并尽可能同时运行它们。

3. 启动事件循环（asyncio.run(main())）：

最后，调用asyncio.run(main())会运行main函数，从而有效地启动事件循环并执行main中安排的任务。

为什么需要这种方法?

整合遗留代码: 在实际应用中，您经常会遇到同步性质的遗留代码。完全重写整个代码库以实现异步兼容性可能是不可行的。通过这种方法，您可以无缝地将这些同步代码集成到异步应用程序中。
与阻塞 I/O 一起工作: 某些操作，特别是涉及阻塞 I/O 的操作，可能没有异步等价物，或者您可能正在使用只提供同步函数的第三方库。这种技术可以将这些操作卸载到线程中，从而释放事件循环来处理其他异步任务。
处理 CPU 密集型任务: 虽然由于 Python 的全局解释器锁 (GIL) 的存在，CPU 密集型任务通常可以通过多进程更好地处理，但有时您可能会选择在线程中运行它们，以简化操作或因为计算开销不会过高。使用run_in_executor允许这些任务与 I/O 绑定的异步任务共存。

Future 对象

在 Python 的异步编程模型中，Future 是一个低级的可等待对象，代表异步操作的最终结果。创建一个 Future 实例时，它是异步结果的一个占位符，将在未来的某个时刻被赋值。Future 是 asyncio 库的重要组成部分，它允许对异步操作进行细粒度控制。

理解 Future

Future 用于连接底层异步操作与高层 asyncio 应用程序。它提供了一种管理异步操作状态的方法：挂起、完成(有结果)或失败(有异常)。
通常在使用高级"异步"函数和结构体(如 Task，它是 Future 的子类)时，不需要自己创建 Future。但了解 Future 对于与低级异步 API 交互或构建复杂异步系统至关重要。

使用 Future

set_result(result): 设置 Future 的结果值。这会将其标记为已完成，并通知所有等待的协程。
set_exception(exception): 用异常作为 Future 的结果值。这也会将其标记为已完成，但等待时会引发该异常。
add_done_callback(callback): 添加回调函数，在 Future 完成(有结果或有异常)时被调用。
result(): 获取 Future 的结果值。如果未完成，将引发 InvalidStateError。如果以异常完成，会重新引发该异常。
done(): 如果 Future 已完成(有结果或有异常)，返回 True。

import asyncio

# 使用 Future 模拟异步操作的函数
async def async_operation(future, data):
    await asyncio.sleep(1)  # 模拟一些有延迟的异步工作
    
    # 根据输入数据设置结果或异常
    if data == "success":
        future.set_result("Operation succeeded")
    else:
        future.set_exception(RuntimeError("Operation failed"))

# Future 完成后调用的回调函数
def future_callback(future):
    try:
        print("Callback:", future.result())  # 尝试打印结果
    except Exception as exc：
        print("Callback:", exc) # 如果有异常，打印异常结果

async def main():
    # 创建一个 Future 对象
    future = asyncio.Future()
    
    # 为 Future 添加回调
    future.add_done_callback(future_callback)
    
    # 启动异步操作并传递 Future
    await async_operation(future, "success") # 尝试将 "success "改成其他名字以模拟失败
    
    # 检查 Future 是否完成并打印其结果
    if future.done():
        try:
            print("Main:", future.result())
        except Exception as exc:
            print("Main:", exc)

asyncio.run(main())

工作原理

async_operation 是一个模拟异步任务的异步函数，接收一个 Future 对象和一些数据(data)作为参数。它会等待1秒钟，模拟异步操作的执行时间. 根据 data 的值，它将使用 set_result 方法在 Future 上设置结果，或使用 set_exception 方法抛出异常.
future_callback 是一个回调函数，在异步操作完成后被调用，用于打印 Future 的结果. 它通过调用 future.result()来获取操作的返回值或重新抛出在 Future 中设置的异常。
在 main 例程中，首先创建一个 Future 对象，并使用 add_done_callback 方法为其添加 future_callback 作为完成回调. 然后调用 async_operation，传入已创建的 Future 对象和样本数据("success"或模拟失败的其他值)。
async_operation 完成后， main 会使用 done() 方法检查 Future 是否已经完成。如果完成，它会尝试直接打印结果；如果遇到异常，则捕获并处理异常。