如何设计和实现一个`Python异步`框架，并解析`asyncio`的`事件循环`、`协程`和`任务调度`。 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

好的，我们开始今天的讲座，主题是设计和实现一个Python异步框架，并深入解析asyncio的事件循环、协程和任务调度。

一、异步编程的核心概念

在深入框架设计之前，我们需要理解异步编程的关键概念：

并发 (Concurrency): 多个任务在一段时间内同时进行。注意，这并不意味着它们真的在同一时刻执行，而是指它们在时间上重叠。
并行 (Parallelism): 多个任务在同一时刻真正地执行。这通常需要多核处理器。
阻塞 (Blocking): 当一个任务等待某个资源时（例如，I/O操作完成），它会停止执行，直到资源可用。
非阻塞 (Non-blocking): 当一个任务等待某个资源时，它不会停止执行，而是立即返回一个状态，指示资源是否可用。
异步 (Asynchronous): 一种非阻塞的并发编程方式，允许程序在等待I/O操作完成时执行其他任务。异步编程通常使用回调、Promise、Future或协程来实现。

二、asyncio 的核心组件

asyncio 是 Python 的标准异步 I/O 库，它提供了构建异步应用的基础设施。其核心组件包括：

事件循环 (Event Loop): asyncio 的心脏。它负责调度任务、处理 I/O 事件和执行回调。
协程 (Coroutine): 一种特殊的函数，可以在执行过程中暂停和恢复。在 asyncio 中，协程使用 async 和 await 关键字定义。
任务 (Task): asyncio.Task 是对协程的包装，用于在事件循环中调度协程的执行。
Future: 代表一个异步操作的最终结果。

三、asyncio 事件循环详解

事件循环是异步编程的核心。它不断地监听 I/O 事件，并调度相应的协程来处理这些事件。

事件循环的生命周期:
- 创建 (Creation): 使用 asyncio.get_event_loop() 创建事件循环。
- 运行 (Running): 使用 loop.run_forever() 或 loop.run_until_complete() 启动事件循环。
- 停止 (Stopping): 使用 loop.stop() 停止事件循环。
- 关闭 (Closing): 使用 loop.close() 关闭事件循环，释放资源。
事件循环的工作原理:

事件循环维护一个就绪队列，其中包含已经准备好运行的协程。它不断地从就绪队列中取出协程并执行，直到队列为空。当一个协程遇到 I/O 操作时，它会暂停执行，并将控制权交还给事件循环。当 I/O 操作完成时，事件循环会将该协程重新添加到就绪队列中。

代码示例:

import asyncio

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

async def main():
    print("Main started")
    await my_coroutine()
    print("Main finished")

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

在这个例子中，asyncio.sleep(1) 是一个协程，它会暂停执行 1 秒钟，并将控制权交还给事件循环。事件循环可以在这段时间内执行其他任务。

四、协程的定义和使用

协程是一种轻量级的线程，可以在执行过程中暂停和恢复。在 asyncio 中，协程使用 async 和 await 关键字定义。

定义协程:

使用 async def 关键字定义协程。

async def my_coroutine():
    print("Hello, world!")

调用协程:

不能直接调用协程。必须使用 await 关键字或 asyncio.create_task() 函数来调度协程的执行。

async def main():
    await my_coroutine()  # 使用 await 调用协程
    task = asyncio.create_task(my_coroutine())  # 创建一个任务
    await task  # 等待任务完成

await 关键字:

await 关键字用于暂停协程的执行，直到一个 awaitable 对象完成。awaitable 对象可以是另一个协程、一个 Future 对象或一个实现了 __await__ 方法的对象。
```
async def fetch_data():
    # 模拟网络请求
    await asyncio.sleep(2)
    return "Data from server"

async def process_data():
    data = await fetch_data()
    print("Processing:", data)

async def main():
    await process_data()

asyncio.run(main())
```
在这个例子中，await asyncio.sleep(2) 会暂停 fetch_data 协程的执行，直到 2 秒钟后。await fetch_data() 会暂停 process_data 协程的执行，直到 fetch_data 协程完成并返回数据。

五、任务的创建和调度

asyncio.Task 是对协程的包装，用于在事件循环中调度协程的执行。

创建任务:

使用 asyncio.create_task() 函数创建一个任务。

async def my_coroutine():
    print("Coroutine started")
    await asyncio.sleep(1)
    print("Coroutine finished")

async def main():
    task = asyncio.create_task(my_coroutine())
    await task  # 等待任务完成

任务的状态:

任务可以处于以下几种状态：
- Pending: 任务正在等待执行。
- Running: 任务正在执行。
- Done: 任务已完成。
- Cancelled: 任务已被取消。

任务的取消:

可以使用 task.cancel() 方法取消任务。

async def my_coroutine():
    try:
        print("Coroutine started")
        await asyncio.sleep(5)
        print("Coroutine finished")
    except asyncio.CancelledError:
        print("Coroutine cancelled")

async def main():
    task = asyncio.create_task(my_coroutine())
    await asyncio.sleep(1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task cancelled")

asyncio.run(main())

在这个例子中，my_coroutine 协程会在 1 秒钟后被取消。CancelledError 异常会在协程中被抛出，允许协程执行清理操作。

六、设计一个简单的异步框架

现在，让我们尝试设计一个简单的异步框架，它将提供以下功能：

任务调度器: 类似于 asyncio.create_task()，用于创建和调度任务。
异步 I/O 支持: 提供异步的网络请求功能。
事件循环管理: 封装事件循环的创建、运行和停止。

框架结构:

my_async_framework/
├── __init__.py
├── event_loop.py
├── task.py
├── http.py

event_loop.py:

import asyncio

class EventLoop:
    def __init__(self):
        self._loop = asyncio.get_event_loop()

    def run_until_complete(self, future):
        self._loop.run_until_complete(future)

    def run_forever(self):
        self._loop.run_forever()

    def stop(self):
        self._loop.stop()

    def close(self):
        self._loop.close()

    def create_task(self, coroutine):
        return self._loop.create_task(coroutine)

    def get_loop(self):
        return self._loop

task.py:

import asyncio

def create_task(coroutine):
    loop = asyncio.get_event_loop()
    return loop.create_task(coroutine)

http.py:

import asyncio
import aiohttp

async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

使用示例:

from my_async_framework.event_loop import EventLoop
from my_async_framework.http import fetch_url
import time

async def main():
    start = time.time()
    tasks = [fetch_url("https://www.example.com") for _ in range(3)]
    results = await asyncio.gather(*tasks) #使用asyncio.gather来并发地执行多个协程。
    end = time.time()
    print(f"Time taken: {end - start:.2f} seconds")
    for result in results:
        print(len(result))

if __name__ == "__main__":
    loop = EventLoop()
    loop.run_until_complete(main())
    loop.close()

这个简单的框架提供了一个事件循环的封装、一个任务创建函数和一个异步的 HTTP 请求函数。

七、框架的进一步扩展

这个框架只是一个起点。可以进一步扩展它，添加以下功能：

更强大的事件循环管理: 支持自定义事件循环策略，例如使用不同的 I/O 多路复用机制。
更灵活的任务调度: 支持任务优先级、任务依赖关系等。
更丰富的异步 I/O 支持: 支持异步的文件 I/O、异步的数据库操作等。
错误处理机制: 提供统一的错误处理机制，例如使用异常处理中间件。
中间件支持: 允许用户自定义中间件来处理请求和响应，例如用于身份验证、日志记录等。

八、asyncio.gather和asyncio.wait 的区别

asyncio.gather 和 asyncio.wait 都是用于并发执行多个协程的工具，但它们在行为上有一些关键的区别。

特性	`asyncio.gather`	`asyncio.wait`
返回值	返回一个列表，包含所有协程的结果，顺序与协程的传入顺序相同。如果任何一个协程引发异常，`gather` 会立即引发该异常，并且取消所有其他正在运行的协程。	返回一个包含两个集合的元组：`done` 集合包含已完成的协程，`pending` 集合包含仍在运行的协程。它不会自动取消其他协程。
异常处理	如果任何一个协程引发异常，`gather` 会立即引发该异常，并且取消所有其他正在运行的协程。	不会引发异常。异常会被包含在已完成的协程的结果中。
取消	如果 `gather` 被取消，所有正在运行的协程也会被取消。	不会自动取消协程。需要手动取消。
使用场景	需要所有协程都成功完成，并且需要按顺序获取结果时。	需要更细粒度的控制，例如在某些协程完成后立即处理它们的结果，或者在发生错误时进行更复杂的处理时。

代码示例:

asyncio.gather:

import asyncio

async def coroutine1():
    await asyncio.sleep(1)
    return "Result from coroutine1"

async def coroutine2():
    await asyncio.sleep(2)
    return "Result from coroutine2"

async def main():
    results = await asyncio.gather(coroutine1(), coroutine2())
    print(results)  # Output: ['Result from coroutine1', 'Result from coroutine2']

asyncio.run(main())

asyncio.wait:

import asyncio

async def coroutine1():
    await asyncio.sleep(1)
    return "Result from coroutine1"

async def coroutine2():
    await asyncio.sleep(2)
    return "Result from coroutine2"

async def main():
    done, pending = await asyncio.wait([coroutine1(), coroutine2()])
    for task in done:
        print(task.result())

asyncio.run(main())

九、总结

今天，我们深入探讨了 Python 异步编程的核心概念，包括事件循环、协程和任务调度。我们还设计了一个简单的异步框架，并讨论了如何进一步扩展它。理解这些概念和技术对于构建高性能、可扩展的异步应用至关重要。选择asyncio.gather 还是 asyncio.wait 取决于你是否需要细粒度的控制，以及如何处理异常。

发表回复 取消回复

发表回复取消回复