Asyncio：入门

定义

asyncio 模块提供了一个围绕事件循环的框架。事件循环基本上是等待某个事件发生，然后对事件作出响应。它负责处理 I/O 和系统事件等事务。

asyncio 实际上提供了多种循环实现。模块会默认选择在当前操作系统下最可能最高效的实现；当然，您也可以显式选择事件循环。事件循环的本质是：“当事件 A 发生时，用函数 B 响应”。

想象一下服务器在等待有人请求某个资源（例如网页）。如果网站不热门，服务器会长时间处于空闲。但一旦有请求进来，服务器就需要做出反应。这种反应称为事件处理。当用户加载网页时，服务器会检查并调用一个或多个事件处理器。一旦这些处理器完成，它们需要把控制权交还给事件循环。要在 Python 中做到这一点，asyncio 使用协程。

协程是一种特殊的函数，它可以在不丢失自身状态的情况下将控制权交还给调用方。协程既是生成器的消费者，也是其扩展。与线程相比，它们的一个重大优势是执行时占用的内存非常少。请注意，当您调用一个协程函数时，它实际上并不会立即执行；它会返回一个协程对象，您可以将其传给事件循环，让它立即或稍后执行。

使用 asyncio 时，您很可能会遇到另一个术语：future。future 基本上是一个对象，代表尚未完成工作的结果。事件循环可以监视 future 对象并等待它们完成。当一个 future 完成时，会被标记为 done。Asyncio 还支持锁和信号量。

最后还要提到的是 Task。Task 是协程的包装器，也是 Future 的子类。您甚至可以使用事件循环来调度 Task。

async 和 await

async 和 await 关键字在 Python 3.5 中被添加，用于定义原生协程，并将其与基于生成器的协程区分为不同的类型。若您想深入了解 async 和 await，请查阅 PEP 492。

在 Python 3.4 中，您会像这样创建协程：

# Python 3.4 coroutine example
# Import `asyncio`
import _______
 
@asyncio.coroutine
def my_coro():
    yield from func()

这个装饰器在 Python 3.5 中仍然可用，但 types 模块新增了一个用于协程函数的类型判断，现在可以告诉您所交互的是不是原生协程。从 Python 3.5 开始，您可以使用 async def 在语法层面定义协程函数。

因此，上面的函数会变成这样：

import _______
 
async def _______():
    await func()

以这种方式定义协程时，您不能在协程函数内使用 yield。相反，必须包含用于向调用方返回值的 return 或 await 语句。请注意，await 关键字只能用于 async def 函数内部。

async/await 关键字可视为用于异步编程的 API。asyncio 模块只是一个框架，恰好使用 async/await 来实现异步编程。实际上有一个名为 curio 的项目证明了这个概念，它是事件循环的另一种实现，在内部使用 async/await。

一个糟糕的协程示例

了解这些工作机制的背景知识固然有帮助，但有时您只想看看一些示例，以便熟悉语法并学会如何把各部分拼在一起。

基于此，我们先从一个简单示例开始！

一个相当常见的任务是从某个位置下载文件，无论是内部资源还是互联网上的文件。通常您会想下载不止一个文件。

那么我们来创建一对可以完成此任务的协程：

import asyncio
import os
import urllib.request
 
async def download_coroutine(url):
    #"A coroutine to download the specified 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):
    """
    Creates a group of coroutines and waits for them to finish
    """
    coroutines = [download_coroutine(url) for url in urls]
    completed, pending = await 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 方法。您会注意到我们将主协程传递给了事件循环。这会启动主协程，随后它会把第二个协程入队并运行起来。这被称为链式协程。

该示例的问题在于它其实并不是协程。原因是 download_coroutine 函数并不异步。问题在于 urllib 不是异步的，并且我也没有使用 await 或 yield。更好的做法是使用 aiohttp 包。我们接着看！

一个更好的协程示例

aiohttp 包专为创建异步 HTTP 客户端和服务器而设计。您可以使用 pip 像这样安装它：

pip install aiohttp技术支持

安装完成后，我们更新代码以使用 aiohttp 来下载文件：

import aiohttp
import asyncio
import async_timeout
import os
 
async def download_coroutine(session, url):
    with async_timeout.timeout(10):
        async with session.get(url) as response:
            filename = __.path.basename(url)
            with open(________, 'wb') as f_handle:
                while True:
                    chunk = await response.content.read(1024)
                    if not chunk:
                        break
                    f_handle.write(_____)
            return await response.release()
 
async def main(loop):
    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"]
 
    async with aiohttp.ClientSession(loop=loop) as session:
        tasks = [download_coroutine(session, url) for url in urls]
        await asyncio.gather(*tasks)
 
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))

您会注意到我们导入了两个新项目：aiohttp 和 async_timeout。后者实际上是 aiohttp 的依赖项之一，允许我们创建超时上下文管理器。我们从代码底部开始往上看。在底部的条件语句中，我们启动异步事件循环并调用主函数。

在主函数中，我们创建一个 ClientSession 对象，并将其传入每个我们想下载的 URL 的下载协程函数中。在 download_coroutine 中，我们创建了一个 async_timeout.timeout() 上下文管理器，它基本上会创建一个 X 秒的计时器。当时间耗尽时，上下文管理器结束或超时。此处超时为 10 秒。接下来我们调用会话的 get() 方法获取响应对象。现在到了有点“魔法”的部分：当您使用响应对象的 content 属性时，它会返回 aiohttp.StreamReader 的实例，使我们能够按任意大小的块下载文件。在读取文件的同时，我们将其写入本地磁盘。最后我们调用响应的 release() 方法，完成响应处理。

根据 aiohttp 的文档，由于响应对象是在上下文管理器中创建的，从技术上讲会隐式调用 release()。但在 Python 中，显式通常优于隐式，且文档中也提示不应仅依赖连接自行关闭，因此在这里我认为最好还是显式释放。

这里仍有一部分是阻塞的，即实际写入磁盘的那段代码。在写文件时，我们依然在阻塞。还有一个名为 aiofiles 的库可以尝试让文件写入也变为异步，但这部分更新就留给读者了。

调度调用

您也可以使用 asyncio 事件循环来调度对常规函数的调用。我们将首先看 call_soon。call_soon 方法基本上会在尽可能快的时候调用您的回调或事件处理器。它以 FIFO 队列方式工作，因此如果某些回调运行时间较长，其他回调就会被延迟，直到之前的回调完成。

我们看一个示例：

import ______
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(_____________, ____))
        print('starting event loop')
        loop.call_soon(functools.partial(_____________, ____, stop=True))
 
        loop.__________()
    finally:
        print('closing event loop')
        loop._____()

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)技术支持

这会将事件处理器的调用延迟一秒，然后调用它，并将循环作为其第一个参数传入。如果您想调度在未来的某个特定时间运行，那么需要获取循环的时间而不是计算机时间。可以这样做：

current_time = loop.time()技术支持

拿到这个时间后，您就可以使用 call_at 函数，把希望调用事件处理器的时间传给它。假设我们希望在五分钟后调用事件处理器，可以这样做：

loop.call_at(current_time + 300, event_handler, loop)技术支持

在这个例子中，我们使用获取到的当前时间并加上 300 秒（五分钟）。这样就可以把事件处理器的调用延迟五分钟！很巧妙！

任务（Tasks）

任务是 Future 的子类，也是协程的包装器。它们使您能够跟踪其何时完成处理。由于它们是一种 Future，其他协程可以等待任务完成，您也可以在任务处理完成时获取它的结果。

我们来看一个简单示例：

import _______

async def my_task(seconds):
    """
    A task to do for a number of seconds
    """
    print('This task is taking {} seconds to complete'.format(
        seconds))
    await asyncio.sleep(_______)
    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._____()
 
    print("The task's result was: {}".format(task_obj.result()))

这里我们创建了一个异步函数，它接收函数运行所需的秒数。这模拟了一个长时间运行的过程。然后我们创建事件循环，通过调用事件循环对象的 create_task 函数创建一个任务对象。create_task 接受我们想转换为任务的函数。接着我们让事件循环运行直到该任务完成。最后，由于任务已经结束，我们获取其结果。

任务也可以很容易通过它们的 cancel 方法取消。在想结束任务时直接调用即可。如果任务在等待其他操作时被取消，任务会抛出 CancelledError。

总结

此时，您应该已经具备自行使用 asyncio 库的入门知识。asyncio 库非常强大，允许您完成许多很酷且有趣的任务。asyncio 库是为网络套接字而设计的。

在 asyncio 之前，支持异步套接字编程的优秀库是 Twisted 框架。另一个有趣的项目是 Dask，它是一个用于分析计算的灵活并行计算库。多查阅 asyncio 模块的文档，让您的灵感尽情迸发吧！