Python标准模块--asyncio

Z时代
2024-01-10
分类：综合

python

asyncio模块作为一个临时的库，在Python 3.4版本中加入。这意味着，asyncio模块可能做不到向后兼容甚至在后续的Python版本中被删除。根据Python官方文档，asyncio通过coroutines、sockets和其它资源上的多路复用IO访问、运行网络客户端和服务端以及其它相关的原始服务等提供了一种单线程并发应用的架构。本文并不能覆盖所有关于asyncio模块的技术点，但是你可以学到如何去使用这个模块，以及为什么它是有用的。

如果你在一些较老的Python版本中需要一些类似于asyncio模块的技术，你可以看Twisted或者gevent。

2 模块使用

2.1 定义

asyncio模块提供了一种关于事件循环的框架。事件循环就是等待一些任务发生，然后执行相应的事件。它也会处理例如IO操作或者系统事件。asyncio实际中有好几种循环实现方式。模块默认使用的方式是其所运行的操作系统上最有效的方式。如果你愿意，你也可以显式地选择其它事件循环方式。一个事件循环就是当事件A发生时，函数B共同起作用。

设想这样一个场景，服务器等待用户访问并请求一些资源，例如网页。如果这个网站不是非常知名的网站，这个服务器将会在很长的时间内处于空闲状态。但是，一旦某个时间用户点击了这个网站，服务器就需要作出响应。这个响应就是事件处理。当一个用户下载网页，服务器将会去检查并调用一个或者多个事件句柄。一旦这些事件句柄完成相应的处理，它们需要将控制交回给事件循环。为了在Python中完成这个任务，asyncio使用协程。

协程是一个特殊的函数，可以将控制交回给它的调用函数，但是并不丢失它的状态。协程是一个消费者函数，并且是生成器的扩展。协程相比线程最大的优势就是执行协程时不需要占用太多内存。你需要注意的是，当你调用一个协程函数，它并没有真正执行。相反，它将会返回一个协程对象，你可以将这个协程对象传递给事件循环，然后可以立即或者稍后执行它。

当你在使用asyncio模块时，另一个你可能会执行的是future。future就是一个可以表示还没有结束的任务结果的对象。你的事件循环可以观察future对象并等待它们结束。当一个future结束时，它被设置为已完成。asyncio模块也支持锁和信号。

本文最后一部分，我将会提到Task。Task是协程的一个框架，是Future的一个子类。你可以在事件循环中对Task进行调度。

2.2 async和await

async和await是Python 3.5中新添加的关键词，用来定义一个原生的协程，以便于和基于协程的生成器相区别。如果你想了解更多关于async和await的知识，你可以去阅读PEP 492。

在Python 3.4中，你可以按照如下方式创建一个协程，

import asyncio
@asyncio.coroutine
def my_foo():
    yield from func()

这个装饰器在Python 3.5中依然有效，但是模块的类型有所更新，协程函数可以告诉你正在交互的是不是一个原生的协程。从Python 3.5开始，你可以使用async def这种语法来定义一个协程函数，所以上述函数可以按照如下方式定义，

import asyncio
async def my_coro():
    await func()

当你以这种方式定义一个协程函数，你不能在函数内部使用yield。取而代之，你必须使用return或者await语句，用于将返回值返回给调用者。你需要注意的是，关键字await只能在async def函数中使用。

关键字async和await可以认为是异步编程中的接口。asyncio模块就是一个可以将async/await用于异步编程的框架。实际上，有一个叫做curio的项目证实了这个概念，那就是它单独实现了在后台使用async/await的事件循环。

2.3 协程示例

尽管上述的描述可以让你获得很多关于协程如何工作的背景知识，有时候，你仅仅想看到一些示例，这样你就可以切身感受到它的语法形式，以及如何将这些代码组合在一起。考虑到这一点，让我们以一个简单的示例开始把。

一个非常常见的任务就是你想完整的下载一个文件，这个文件可能来源于内部资源或者互联网。当然你想要下载的文件可能不止一个。让我们创建两个协程来完成这个任务。

import asyncio
import os
import urllib.request
 async def download_coroutine(url):
    request = urllib.request.urlopen(url)
    filename = os.path.basename(url)
    with open(filename,"wb") as file_handle:
        while True:
            chunk = request.read(1024)
            if not chunk:
                break
            file_handle.write(chunk)
        msg = "Finished downloading {filename}".format(filename = filename)
        return msg
async def main(urls):
    coroutines = [download_coroutine(url) for url in urls]
    completed,pending = awit asyncio.wait(coroutines)
    for item in completed:
        print(item.result())
if __name__ == "__main__":
    urls = ["http://www.irs.gov/pub/irs-pdf/f1040.pdf",
            "http://www.irs.gov/pub/irs-pdf/f1040a.pdf",
            "http://www.irs.gov/pub/irs-pdf/f1040ez.pdf",
            "http://www.irs.gov/pub/irs-pdf/f1040es.pdf",
            "http://www.irs.gov/pub/irs-pdf/f1040sb.pdf"]
    event_loop = asyncio.get_event_loop()
    try:
        event_loop.run_until_complete(main(urls))
    finally:
        event_loop.close()

这段代码中，我们引入了我们需要的模块，然后通过async语法创建了第一个协程。这个协程叫做download_coroutine，它使用Python的urllib模块下载传递给它的任何URL地址。当它完成任务时，它将会返回一条相应的信息。

另一个协程就是我们的主协程。它基本上就是获取一个包含一个或者多个URL地址的列表，然后将它们加入队列。我们使用asyncio的wait函数用于等待协程的结束。当然，为了启动这些协程，它们需要被加入到事件循环中。我们在代码段中最后的地方做了这个处理，我们先获取一个事件循环，然后调用它的run_until_complete的方法。你将会注意到，我们将主协程传入事件循环中。这个会先运行主协程，主协程将第二个协程加入到队列中，并让它们运行。这就是有名的链协程。

2.4 调度调用

你也可以通过异步事件循环来调度调用常规函数。我们看的第一个方法是call_soon。方法call_soon基本上就是尽可能的调用你的回调或者事件句柄。它的工作机制类似于先进先出队列，所以如果一些回调需要一段时间来处理任务，其它的回调就会相应的延迟，直到先前的回调结束。让我们来看一个示例。

import asyncio
import functools
def event_handler(loop,stop = False):
    print("Event handler called")
    if stop:
        print("Stopping the loop")
        loop.stop()
if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    try:
        loop.call_soon(functools.partial(event_handler,loop))
        print("Starting event loop")
        loop.call_soon(functools.partial(event_handler,loop,stop = True))
        loop.run_forever()
    finally:
        print("closing event loop")
        loop.close()

由于asyncio的函数不接受关键字，但是如果我们需要将关键字传入事件句柄中，那么我们就需要使用functools模块了。无论何时被调用，我们定义的常规函数将会在标准输出上打印一些文字信息。如果你偶然将这个函数的stop变量设置为True，它将会停止事件循环。

第一次我们调用它时，我们没有停止事件循环。第二次我们调用它时，我们停止了事件循环。我们停止事件循环的原因是我们将它放入run_forever，这个将时间循环设置为无限循环。一旦循环停止，我们就可以将它关闭。如果你运行这段代码，你得到的输出如下所示，

Starting event loop Event handler called Event handler called Stopping the loop closing event loop

还有一个相关的函数是call_soon_threadsafe，顾名思义，它与call_soon的工作机制相似，但是它是线程安全的。

如果你想延迟一段时间再调用，你可以使用call_later函数。在这个示例中，我们可以将call_soon函数按照如下方式修改，

loop.call_later(1,event_handler,loop)

这个将会延迟调用我们的事件句柄1秒钟，然后才会去调用它，并将循环作为第一个参数传入。

如果你想在未来一个指定的时间调度，你需要获取循环的时间，而不是计算机的时间，你可以按照如下方式操作，

current_time = loop.time()

一旦你这样做，你可以使用call_at函数，然后将你想调用事件句柄的时间传递给它。让我们来看看我们想在5分钟之后调用我们的事件句柄，下面就是你如何操作的，

loop.call_at(current_time + 300,event_handler,loop)

在这个示例中，我们使用我们获取的当前时间，然后加上300秒钟或者5分钟。通过这个操作，我们延迟调用事件循环5分钟。

2.5 任务

Task是Future的一个子类，也是协程的一个框架。Task可以让你记录到任务结束处理的时间。由于任务是Future类型，其它的协程可以等待一个任务，当任务处理完毕时你也可以获取到它的结果。让我们看一个简单的示例。

import asyncio
import time
async def my_task(seconds):
    print("This task is take {} seconds to cpmplete".format(seconds))
    time.sleep(seconds)
    return "task finished"
if __name__ == "__main__":
    my_event_loop = asyncio.get_event_loop()
    try:
        print("task creation started")
        task_obj = my_event_loop.create_task(my_task(seconds = 2))
        my_event_loop.run_until_complete(task_obj)
    finally:
        my_event_loop.close()
    print("The task's result was :{}".format(task_obj.result()))

在这里，我们创建一个异步函数，它接受秒数，也是它将会运行的时间。这个模仿了一个长时间运行的任务。然后我们创建了我们的事件循环，并且通过事件循环对象的create_task函数创建了一个任务对象。函数create_task接受我们想要转换为任务的函数。然后我们运行事件循环，直到任务完成。在最后，一旦任务结束，我们就获得任务的结果。

通过任务的cancel方法，任务也可以很容易被取消。当你想结束一个任务，调用它就可以了。当一个任务在等待另一个操作时被取消，这个任务将会报出CancelError错误。

2.6 总结

到这里，你应该已经了解如何利用asyncio库进行工作了。asyncio库是非常强大的，它允许你去做很多酷并且有意思的任务。你可以查看http://asyncio.org/，该网站包含了很多使用asyncio的项目，可以获取到很多关于如何使用asyncio库的灵感。当然，Python官方文档也是一个很好的开始asyncio之旅的地方。