Python:从头创建 Asyncio (2)

简介: Python:从头创建 Asyncio (2)

引言

现在,asyncio 已成为 Python 社区中的热门话题,并且名副其实——它提供了一种非常出色的处理 I/O 密集型程序的方法!在我探索 asyncio 的过程中,我起初并不太明白它的工作原理。但随着深入学习,我意识到 asyncio 实际上是在 Python 生成器的基础上增加了一层非常便利的封装。

本文中,我将展示如何仅用 Python 生成器来构建一个 asyncio 的简化模型。接着,我会演示如何利用 await 魔法方法,将示例代码改写为使用 async 和 await 关键字。最终,我会将我的简化版本替换为官方的 asyncio 库。通过这个过程,我相信你将对 asyncio 的神奇之处有一个更深入的理解。

Sleeping

如果我们沿用之前示例中的代码,我们可以通过 yield from 的应用,为我们的任务嵌入子生成器。例如,我在这里引入了一个休眠生成器,它会在指定的时间到达之前暂停任务的执行。这种机制之所以有效,是因为 sleep 函数会连续产生 yield,直到经过了设定的秒数,然后它将跳出 while 循环。

由于 sleep 函数中没有其他 yield 语句,这将引发一个 StopIteration 异常,这个异常告诉 yield from 语句在任务函数中跳过当前的生成器,继续执行下一行代码。

import time

def sleep(seconds):
   start_time = time.time()
   while time.time() - start_time < seconds:
       yield

def task1():
   while True:
       print('Task 1')
       yield from sleep(1)

def task2():
   while True:
       print('Task 2')
       yield from sleep(5)

event_loop = [task1(), task2()]

while True:
   for task in event_loop:
       next(task)

输出:

Task 1
Task 2
Task 1
Task 1
Task 1
Task 1
Task 2
Task 1
…

Yield to Await

我们现在可以将之前的代码示例,通过应用 _await__ 魔术方法和 async 关键字,从使用 yield 转变为使用 await。如果一个类定义了 _await__ 方法,我们就可以在该类的实例前加上 await 关键字来调用这个方法。在 asyncio 框架中,你通常通过调用如 asyncio.create_task 这样的函数来处理 Task 对象。这些 Task 对象是从 asyncio 的 Future 对象派生而来的,而 Future 对象定义了 _await__ 方法。我们还可以在协程前使用 await,协程是在函数定义时加上 async 关键字生成的对象。协程和生成器函数类似,它们的执行都能够被挂起和恢复。

你可以将 await 关键字理解为 yield from 的一个变体,它附带了一些额外的验证规则。因此,当你在代码中写 await object 时,你实际上是在指示从 "object" 类的实例中调用 _await__ 方法,或者 "object" 本身可能就是另一个协程(类似于子生成器)。

实际上,你甚至可以查看 Asyncio 的源代码,发现 Future 对象中的 _await__ 方法在调用时,如果未来(或任务)尚未完成,它基本上只是执行了 yield 操作。

要将我们在上一节中编写的代码转移到使用 async 和 await,我们首先需要创建自己的 Task 类,因为函数不能具有 await dunder 方法。下面是我想出的一个简单版本:

from queue import Queue


event_loop = Queue()

class Task():
    def __init__(self, generator):
        self.iter = generator
        self.finished = False

    def done(self):
        return self.finished

    def __await__(self):
        while not self.finished:
            yield self


def create_task(generator):
    task = Task(generator)
    event_loop.put(task)

    return task

这一次,我们改用队列而非 Python 列表来构建事件循环,这样做更合理,因为我们希望添加或移除任务的操作能够快速完成,即在常数时间内完成。

在我们的 Task 类中,我们将生成器对象保存在 self.iter 属性中,并设置 self.finished 属性为 False,用以跟踪生成器是否已经运行完毕(当生成器引发 StopIteration 异常时,表示其运行结束)。Task 对象还定义了一个 await 魔术方法,这个方法将持续地将控制权交还给事件循环,直到任务完成。完成 Task 对象的创建后,我们使用 create_task 辅助函数将它加入到事件循环中,这将安排它按计划执行。

接下来,我们将构建事件循环管理器,它负责驱动任务的执行。

def run(main):
    event_loop.put(Task(main))

    while not event_loop.empty():
        task = event_loop.get()
        try:
            task.iter.send(None)
        except StopIteration:
            task.finished = True
        else:
            event_loop.put(task)

你或许已经发现,我们的实现开始接近 asyncio 的实际 API。要启动事件循环,我们需要通过一个初始函数来调用 run。这个函数首先将主函数封装进 Task 对象,并加入到事件循环中。随后,while 循环会启动,并且在每次迭代中,通过队列来获取下一个待执行的任务。现在我们使用 task.iter.send(None) 替代了 next(task.iter),这在使用 async/await 关键字时显得有些奇特,但功能上是一致的。我们还需要将这个调用放在 try-except 块中,以便在抛出 StopIteration 异常时,可以将 task.finished 设置为 True;如果没有异常抛出,代码将执行 else 语句,这会把任务重新放回事件循环中,以便再次执行。

接下来,我们需要对 sleep 函数进行异步兼容改造。之前,我们通过一个带有 while 循环和单个 yield 的生成器函数来实现休眠功能。尽管我偏爱这种方法,但 await 关键字不能与生成器函数一起使用——它需要是一个定义了 await 魔术方法的对象或是一个协程函数。因此,为了解决这个问题,我将代码迁移到了另一个函数中,现在实际的 sleep 函数会创建一个任务对象并等待它完成。这个 await 调用将触发 Task 对象内的 await 方法,随后执行 yield,允许事件循环转向其他任务。当事件循环处理到新的 _sleep 任务时,它会检查时间,如果时间未到,同样会执行 yield,将控制权交还给事件循环。如果休眠的任务再次被事件循环调用,就像生成器保存其状态一样,协程仍在等待 sleep 函数返回。由于 sleep 函数还在等待 _sleep 任务完成,任务的 await 魔术方法将再次被调用,由于任务尚未结束,魔术方法中的 yield 将再次被执行。

import time

def _sleep(seconds):
    start_time = time.time()
    while time.time() - start_time < seconds:
        yield


async def sleep(seconds):
    task = create_task(_sleep(seconds))
    return await task

以下是所有代码的汇总:

from queue import Queue
import time


event_loop = Queue()


def _sleep(seconds):
    start_time = time.time()
    while time.time() - start_time < seconds:
        yield


async def sleep(seconds):
    task = create_task(_sleep(seconds))
    return await task


class Task():
    def __init__(self, generator):
        self.iter = generator
        self.finished = False

    def done(self):
        return self.finished

    def __await__(self):
        while not self.finished:
            yield self


def create_task(generator):
    task = Task(generator)
    event_loop.put(task)

    return task


def run(main):
    event_loop.put(Task(main))

    while not event_loop.empty():
        task = event_loop.get()
        try:
            task.iter.send(None)
        except StopIteration:
            task.finished = True
        else:
            event_loop.put(task)

既然我们已经成功构建了事件循环、任务创建机制和 sleep 函数,接下来我们可以引入名为 "jacobio.py" 的文件,并把之前使用 yield 语句的部分替换成 await 调用。同时,我们需要在那些使用了 await 的函数前加上 async 关键字,以表明这些函数是异步的,并且可以被其他代码等待执行。最后,我们还需要像在 asyncio 库中那样编写一个主函数,用于将任务排入事件循环的执行队列中。

import jacobio

async def task1():
    for _ in range(2):
        print('Task 1')
        await jacobio.sleep(1)

async def task2():
    for _ in range(3):
        print('Task 2')
        await jacobio.sleep(0)

async def main():
    one = jacobio.create_task(task1())
    two = jacobio.create_task(task2())

    await one
    await two

    print('done')


if __name__ == '__main__':
    jacobio.run(main())

输出:

Task 1
Task 2
Task 2
Task 2
Task 1
done

Await with AsyncIO

现在,我们可以从上面获取代码,并将所有出现的“jacobio”替换为“asyncio”,我们现在完全使用 asyncio 包!

import asyncio

async def task1():
    for _ in range(2):
        print('Task 1')
        await asyncio.sleep(1)

async def task2():
    for _ in range(3):
        print('Task 2')
        await asyncio.sleep(0)

async def main():
    one = asyncio.create_task(task1())
    two = asyncio.create_task(task2())

    await one
    await two

    print('done')


if __name__ == '__main__':
    asyncio.run(main())

Asyncio 在后台执行了许多复杂的操作,但我们成功地从基础的生成器出发,一步步重建了 asyncio 的核心功能!我努力使事件循环管理器的设计尽可能简洁,尽管这仅是 asyncio 工作理念的简化版,与实际的库相比,我的实现在细节上与官方源代码的执行流程有所不同。此外,既然我们现在拥有了完整的 asyncio 库的功能,就无需为了同时等待两个任务而分别创建它们;我们完全可以使用 asyncio.gather() 这样的函数来同时管理多个任务。

相关文章
|
2月前
|
搜索推荐 程序员 调度
精通Python异步编程:利用Asyncio与Aiohttp构建高效网络应用
【10月更文挑战第5天】随着互联网技术的快速发展,用户对于网络应用的响应速度和服务质量提出了越来越高的要求。为了构建能够处理高并发请求、提供快速响应时间的应用程序,开发者们需要掌握高效的编程技术和框架。在Python语言中,`asyncio` 和 `aiohttp` 是两个非常强大的库,它们可以帮助我们编写出既简洁又高效的异步网络应用。
162 1
|
1月前
|
Python
Python中的异步编程:使用asyncio和aiohttp实现高效网络请求
【10月更文挑战第34天】在Python的世界里,异步编程是提高效率的利器。本文将带你了解如何使用asyncio和aiohttp库来编写高效的网络请求代码。我们将通过一个简单的示例来展示如何利用这些工具来并发地处理多个网络请求,从而提高程序的整体性能。准备好让你的Python代码飞起来吧!
75 2
|
1月前
|
调度 开发者 Python
Python中的异步编程:理解asyncio库
在Python的世界里,异步编程是一种高效处理I/O密集型任务的方法。本文将深入探讨Python的asyncio库,它是实现异步编程的核心。我们将从asyncio的基本概念出发,逐步解析事件循环、协程、任务和期货的概念,并通过实例展示如何使用asyncio来编写异步代码。不同于传统的同步编程,异步编程能够让程序在等待I/O操作完成时释放资源去处理其他任务,从而提高程序的整体效率和响应速度。
|
28天前
|
API 调度 开发者
探索Python中的异步编程:从asyncio到Trio
本文将带你深入Python异步编程的心脏地带,从asyncio的基本概念到Trio的高级特性,我们将一起揭开Python异步编程的神秘面纱,并探讨它们如何改变我们的编程方式。
|
27天前
|
API 开发者 Python
探索Python中的异步编程:Asyncio与Tornado的对决
在这个快节奏的世界里,Python开发者面临着一个挑战:如何让代码跑得更快?本文将带你走进Python异步编程的两大阵营——Asyncio和Tornado,探讨它们如何帮助我们提升性能,以及在实际应用中如何选择。我们将通过一场虚拟的“对决”,比较这两个框架的性能和易用性,让你在异步编程的战场上做出明智的选择。
|
1月前
|
数据库 Python
异步编程不再难!Python asyncio库实战,让你的代码流畅如丝!
在编程中,随着应用复杂度的提升,对并发和异步处理的需求日益增长。Python的asyncio库通过async和await关键字,简化了异步编程,使其变得流畅高效。本文将通过实战示例,介绍异步编程的基本概念、如何使用asyncio编写异步代码以及处理多个异步任务的方法,帮助你掌握异步编程技巧,提高代码性能。
74 4
|
1月前
|
API 数据处理 Python
探秘Python并发新世界:asyncio库,让你的代码并发更优雅!
在Python编程中,随着网络应用和数据处理需求的增长,并发编程变得愈发重要。asyncio库作为Python 3.4及以上版本的标准库,以其简洁的API和强大的异步编程能力,成为提升性能和优化资源利用的关键工具。本文介绍了asyncio的基本概念、异步函数的定义与使用、并发控制和资源管理等核心功能,通过具体示例展示了如何高效地编写并发代码。
42 2
|
28天前
|
测试技术 Python
Python中的异步编程与`asyncio`库
Python中的异步编程与`asyncio`库
|
1月前
|
调度 开发者 Python
异步编程在Python中的应用:Asyncio和Coroutines
异步编程在Python中的应用:Asyncio和Coroutines
29 1
|
2月前
|
关系型数据库 MySQL 数据处理
探索Python中的异步编程:从asyncio到异步数据库操作
在这个快节奏的技术世界里,效率和性能是关键。本文将带你深入Python的异步编程世界,从基础的asyncio库开始,逐步探索到异步数据库操作的高级应用。我们将一起揭开异步编程的神秘面纱,探索它如何帮助我们提升应用程序的性能和响应速度。