异步编程

在开始说明异步编程之前，首先先了解几个相关的概念。

 

阻塞

程序未得到所需计算资源时被挂起的状态。

程序在等待某个操作完成期间，自身无法继续干别的事情，则称该程序在该操作上是阻塞的。

常见的阻塞形式有：网络I/O阻塞、磁盘I/O阻塞、用户输入阻塞等。

阻塞是无处不在的，包括CPU切换上下文时，所有的进程都无法真正干事情，它们也会被阻塞。（如果是多核CPU则正在执行上下文切换操作的核不可被利用。）

简单的理解的话，阻塞就是 A调用B，A会被挂起，一直等待B的结果，什么都不能干。

 

非阻塞

程序在等待某操作过程中，自身不被阻塞，可以继续运行干别的事情，则称该程序在该操作上是非阻塞的。

非阻塞并不是在任何程序级别、任何情况下都可以存在的。

仅当程序封装的级别可以囊括独立的子程序单元时，它才可能存在非阻塞状态。

非阻塞的存在是因为阻塞存在，正因为某个操作阻塞导致的耗时与效率低下，我们才要把它变成非阻塞的。

简单理解的话，非阻塞就是 A调用B，A自己不用被挂起来等待B的结果，A可以去干其他的事情

 

同步

不同程序单元为了完成某个任务，在执行过程中需靠某种通信方式以协调一致，称这些程序单元是同步执行的。

例如购物系统中更新商品库存，需要用“行锁”作为通信信号，让不同的更新请求强制排队顺序执行，那更新库存的操作是同步的。

简言之，同步意味着有序。

简单理解的话，同步就是A调用B，此时只有等B有了结果才返回

 

异步

为完成某个任务，不同程序单元之间过程中无需通信协调，也能完成任务的方式。

不相关的程序单元之间可以是异步的。

例如，爬虫下载网页。调度程序调用下载程序后，即可调度其他任务，而无需与该下载任务保持通信以协调行为。不同网页的下载、保存等操作都是无关的，也无需相互通知协调。这些异步操作的完成时刻并不确定。

简言之，异步意味着无序。

简单理解的话，异步就是A调用B，B立即返回，无需等待。等B处理完之后再告诉A结果

 

并发

并发描述的是程序的组织结构。指程序要被设计成多个可独立执行的子任务。

以利用有限的计算机资源使多个任务可以被实时或近实时执行为目的。

 

并行

并行描述的是程序的执行状态。指多个任务同时被执行。

以利用富余计算资源（多核CPU）加速完成多个任务为目的。

并发提供了一种程序组织结构方式，让问题的解决方案可以并行执行，但并行执行不是必须的。

总的来说，并行是为了利用多核加速多任务的完成；并发是为了让独立的子任务能够尽快完成；非阻塞是为了提高程序的整体运行效率，而异步是组织非阻塞任务的方式。

 

并发

是指的程序的组织结构，把程序设计成多个可以独立执行的子任务。目的是使用有限的计算机资源使得多个任务可以被实时或者接近实时执行为目的。

 

并行

指的是程序的执行状态，多个任务同时执行。这样做的目的是为了利用富余的计算资源（多核cpu），加速完成多个任务的目的。

并发提供了一种组织结构方式，让问题的解决方式可以并行执行，但是这不是必须的。

 

综上

并行是为了利用多核计算机的富余计算资源来加速多任务程序完成的进度。

并发是为了让独立的子任务有机会被尽快执行，但是不一定会加快整体的进度。

非阻塞是为了提高程序执行的整体效率。

异步是组织非阻塞任务的方式。

 

异步编程

以进程、线程、协程、函数/方法作为执行任务程序的基本单位，结合回调、事件循环、信号量等机制，以提高程序整体执行效率和并发能力的编程方式。

异步执行的程序一定是无序的，如果你可以根据已经执行的指令，准确的判断出，它接下里要去执行的某个具体操作，那么它就是同步程序。这是有序和无序的区别。

异步编程的困难在于，因为异步执行的程序，它的执行顺序不可预料，所以在并行情况下变得比较复杂和艰难。

所以几乎所有的异步框架都将异步编程模型简化：一次只允许处理一个事件。因此关于异步的讨论基本上都集中在了单线程中。

如果某个事件的处理过程过长，那么其他部分就会被阻塞。

所以异步编程的异步调度必须要“足够小”，不能耗时太久。

合理的用异步编程的方式可以提高 cpu 的利用率，提高程序效率。

 

同步->异步I/O

以一个爬虫为例，下载10篇网页，用几个例子来展示从同步->异步。

 

同步阻塞方式

以同步阻塞方式来写这个程序也是最容易想到的方式，即依次下载好10篇网页。

import socket



def blocking_way():

    sock = socket.socket()

    # 阻塞

    sock.connect(('example.com', 80))

    request = 'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n'

    sock.send(request.encode('ascii'))

    response = b''

    chunk = sock.rev(4096)

    while chunk:

        response += chunk

        # 阻塞

        chunk = sock.rev(4096)

    return response



def sync_way():

    res = []

    for i in range(10):

        res.append(blocking_way())

    return len(res)

复制代码

 

这段代码的执行事件大概为4.5秒。（取多次平均值）

上述代码中， blocking_way()这个函数的作用主要是建立连接，发送HTTP请求，然后从socket读取HTTP响应请求到并返回数据。

sync_way()将blocking_way()执行了10次，也就是说，我们执行了10次访问下载 example.com

由于网络情况和服务端的处理各不相同，所以服务端什么时候返回了响应数据并被客户端接收到可供程序读取，也是不可预测的。所以 sock.connect() 和 sock.recv() 这两个调用在默认情况下是阻塞的。

注：sock.send()函数并不会阻塞太久，它只负责将请求数据拷贝到TCP/IP协议栈的系统缓冲区中就返回，并不等待服务端返回的应答确认。

如果是说网络环境很差的话，创建网络连接的TCP/IP握手需要1秒，那么 sock.connect() 就得阻塞1秒。这一秒时间对CPU来说就被浪费了。同理，sock.recv() 也一样的必须得等到服务端的响应数据已经被客户端接收，才能进行后续的程序。目前的例子上只有只需要下载一篇网页，阻塞10次看起来好像没有什么问题，可是如果需求是1000w篇的话，这种阻塞的方式就显得很蠢，效率也很低下。

 

改进：多进程

在一个程序中，依次执行10次好像有些耗时，那么我们使用多进程，开10个同样的程序一起处理的话，也许会好一些？于是第一个改进方式便出来了：多进程编程。发展脉络也是如此。在更早的操作系统（Linux 2.4）及其以前，进程是 OS 调度任务的实体，是面向进程设计的OS。

import socket

from concurrent import futures



def blocking_way():

    sock =socket.socket()

    # 阻塞

    sock.connect(('example.com', 80))

    request = 'GET /HTTP/1.0\r\nHost: example.com\r\n\r\n'

    sock.send(request.encode('acsii'))

    response = b''

    chunk = sock.recv(4096)

    while chunk:

        response += chunk

        chunk = sock.recv(4096)

    return response



def process_way():

    workers = 10

    with futures.ProcessPoolExecutor(workers) as executor:

    futs = {executor.sumbit(blocking_way) for i in range(10)}

    return len([fut.result() for fut in futs])

复制代码

 

这段代码执行时间大概为0.6秒。

按理说，使用10个相同的进程来执行这段程序，其执行时间应该是会缩短到原来的1/10，然而并没有。这里面还有一些时间被进程的切换所消耗掉了。

CPU从一个进程切换到另一个进程的时候，需要把旧进程运行时的寄存器状态，内存状态都保存好，然后再将另一个进程之前保存的数据恢复。当进程数量大于CPU核心数的时候，进程切换是必须的。

一般来说，服务器在能够稳定运行的前提下，可以同时处理的进程数在数十个到数百个规模。如果进程数量规模更大，系统运行将不稳定，而且可用内存资源往往也会不足。除了切换开销大，以及可支持的任务规模小之外，多进程还有其他缺点，如状态共享等问题。

 

改进：多线程

线程的数据结构比进程更加的轻量级，同一个进程可以容纳好几个线程。

后来的OS也把调度单位由进程转为线程，进程只作为线程的容器，用于管理进程所需的资源。而且OS级别的线程是可以被分配到不同的CPU核心同时运行的。

import socket

from concurrent import futures





def blocking_way():

    sock = socket.socket()

    # 阻塞

    sock.connect(('example.com', 80))

    request = 'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n'

    sock.send(request.encode('acsii'))

    response = b''

    chunk = sock.recv(4096)

    while chunk:

        response += chunk

        # 阻塞

        chunk = sock.recv(4096)

    return response



def thread_way():

    wokers = 10

    with futures.ThreadPoolExecutor(workers) as executor:

        futs = {executor.sumbit(blocking_way) for i in range(10)}

    return len([fut.result(fut.result() for fut in futs)])

复制代码

 

总运行时间大概为0.43秒。

从运行时间上来看，多线程好像已经解决了进程切换开销大的问题，而且可支持的任务数量规模，也变成了数百个到数千个。

但是由于CPython中的多线程因为GIL的存在，它们并不能利用CPU多核优势，一个Python进程中，只允许有一个线程处于运行状态。

在做阻塞的系统调用时，例如sock.connect(),sock.recv()时，当前线程会释放GIL，让别的线程有执行机会。但是单个线程内，在阻塞调用上还是阻塞的。

Python中 time.sleep 是阻塞的，都知道使用它要谨慎，但在多线程编程中，time.sleep 并不会阻塞其他线程。

除了GIL之外，所有的多线程还有通病。它们是被OS调度，调度策略是抢占式的，以保证同等优先级的线程都有均等的执行机会，那带来的问题是：并不知道下一时刻是哪个线程被运行，也不知道它正要执行的代码是什么。所以就可能存在竞态条件。如果在一个复杂的爬虫系统中，要抓取的URL由多个爬虫线程来拿，那么URL如何分配，这就需要用到“锁”或“同步队列”来保证下载任务不会被重复执行。多线程最主要的问题还是竞态条件。

 

非阻塞方式

千呼万唤使出来，下例是最原始的非阻塞。

import socket



def noblock_way():

    sock = socket.socket()

    sock.setblocking(False)



    try:

        sock.connect(('example.com', 80))

    except BlockingIOError:

        # 非阻塞过程也会抛出异常

        pass

    

    request = 'GET / HTTP /1.0\r\nHost: example.com\r\n\r\n'

    data = request.encode('ascii')

    # 不断重复尝试发送

    while True:

        try:

            sock.send(data)

            # send不出现异常，停止

            break

        except OSError:

            pass



    response = b''

    while True:

        try:

            chunk = sock.recv(4096)

            while chunk:

                response += chunk

                chunk = sock.recv(4096)

            break

        except OSError:

            pass

    

    return response



def sync_way():

    res = []

    for i in range(10):

        res.append(noblock_way())

    return len(res)

复制代码

 

程序总耗时约4.3秒。

执行完这段代码的时候，感觉好像是被骗了，代码的执行时间和非阻塞方式差不多，而且程序更复杂了。要非阻塞何用？

代码sock.setblocking(False)告诉OS，让socket上阻塞调用都改为非阻塞的方式。非阻塞就是在做一件事的时候，不阻碍调用它的程序做别的事情。上述代码在执行完 sock.connect() 和 sock.recv() 后的确不再阻塞，可以继续往下执行请求准备的代码或者是执行下一次读取。第8行要放在try语句内，是因为socket在发送非阻塞连接请求过程中，系统底层也会抛出异常。connect()被调用之后，立即可以往下执行第12和13行的代码。

虽然 connect() 和 recv() 不再阻塞主程序，空出来的时间段CPU没有空闲着，但并没有利用好这空闲去做其他有意义的事情，而是在循环尝试读写 socket （不停判断非阻塞调用的状态是否就绪）。还得处理来自底层的可忽略的异常。也不能同时处理多个 socket。 所以总体执行时间和同步阻塞相当。

 

非阻塞改进

 

epoll

其实判断非阻塞调用是否就绪可以交给OS来做，不用应用程序自己去等待和判断，可以用这个空闲时间去做其他的事情。

OS将O/I的变化都封装成了事件，比如可读事件、可写事件。而且提供了相应的系统模块以供调用来接收事件通知。这个模块就是select，让应用程序可以通过select注册文件描述符和回调函数。当文件描述符的状态发生变化时，select 就调用事先注册的回调函数。

select因其算法效率比较低，后来改进成了poll，再后来又有进一步改进，BSD内核改进成了kqueue模块，而Linux内核改进成了epoll模块。这四个模块的作用都相同，暴露给程序员使用的API也几乎一致，区别在于kqueue 和 epoll 在处理大量文件描述符时效率更高。一般的Linux服务器是使用的 epoll。

 

回调（callback)

将I/O事件的监听交给OS来处理，那么OS在知道I/O状态发生改变之后应该如何处理呢，这里一般都是通过回调的方式。

把发送数据和读取数据封装成独立的函数，用epoll代替应用程序监听socket状态，而且需要告知epoll “如果socket状态变为可以往里写数据（连接建立成功了），请调用HTTP请求发送函数。如果socket 变为可以读数据了（客户端已收到响应），请调用响应处理函数。”

import socket

from selectors import DefaultSelector, EVENT_WRITE, EVENT_READ

# selectors模块是对底层select/poll/epoll/kqueue的封装

# DefaultSelector类会根据 OS 环境自动选择最佳的模块



"""

创建Crawler 实例；

调用fetch方法，会创建socket连接和在selector上注册可写事件；

fetch内并无阻塞操作，该方法立即返回；

重复上述3个步骤，将10个不同的下载任务都加入事件循环；

启动事件循环，进入第1轮循环，阻塞在事件监听上；

当某个下载任务EVENT_WRITE被触发，回调其connected方法，第一轮事件循环结束；

进入第2轮事件循环，当某个下载任务有事件触发，执行其回调函数；此时已经不能推测是哪个事件发生，因为有可能是上次connected里的EVENT_READ先被触发，也可能是其他某个任务的EVENT_WRITE被触发；（此时，原来在一个下载任务上会阻塞的那段时间被利用起来执行另一个下载任务了）

循环往复，直至所有下载任务被处理完成

退出事件循环，结束整个下载程序

"""



selector = DefaultSelector()

stopped = False

urls_todo = {'/', '/1', '/2', '/3', '/4', '/5', '/6', '/7', '/8', '/9'}





class Crawler:

    """

    如果用这种方法抓去，需要创建10个Crawler实例，这样就会有20个事件发生

    """

    def __init__(self, url):

        self.url = url

        self.sock = None

        self.response = b''



    def fetch(self):

        self.sock = socket.socket()

        self.sock.setblocking(False)

        try:

            self.sock.connect(('example.com', 80))

        except BlockingIOError:

            pass

        selector.register(self.sock.fileno(), EVENT_WRITE, self.connected)



    def connected(self, key, mask):

        selector.unregister(key.fd)

        get = 'GET {0} HTTP/1.0\r\nHost: example.com\r\n\r\n'.format(self.url)

        self.sock.send(get.encode('ascii'))

        selector.register(key.fd, EVENT_READ, self.read_response)



    def read_response(self, key, mask):

        global stopped

        # 如果响应大于4kb，下次循环继续

        chunk = self.sock.recv(4096)

        if chunk:

            self.response += chunk

        else:

            selector.unregister(key.fd)

            urls_todo.remove(self.url)

            if not urls_todo:

                stopped =True



def loop():

    while not stopped:

        # 阻塞，直到一个事件发生

        events = selector.select()  # 这是一个阻塞调用

        for event_key, event_mask in events:

            callback = event_key.data

            callback(event_key, event_mask)



if __name__ == "__main__":

    import time

    start = time.time()

    for url in urls_todo:

        crawler = Crawler(url)

        crawler.fetch()

    loop()

    print(time.time() - start)

复制代码

 

总体耗时约0.45秒。

与之前函数不太一眼的地方是，我们将下载10个不同的URL界面，然后将URL的相对路径存储在 urls_todo 中。具体的改进如下。

首先是不断尝试 send() 和 recv() 这两个循环被取消掉了。

其次，导入了selectors模块，并创建了一个DefaultSelector 实例。Python标准库提供的selectors模块是对底层select/poll/epoll/kqueue的封装。DefaultSelector类会根据 OS 环境自动选择最佳的模块，那在 Linux 2.5.44 及更新的版本上都是epoll了。

然后分别注册了socket可写事件(EVENT_WRITE)以及可读事件(EVENT_READ)发生后应该采取的回调函数。

但是这里有一个问题，我们如何才能知道这10个Crawler实例创建的20个事件，哪个是当前正在发生的事件，从selector中拿出来，并且得到对应的回调函数去执行呢？

 

事件循环

所以我们在代码结尾加入了事件循环，写一个函数，循环地去访问selector模块，等待它告诉我们当前是哪个事件发生了，对应的应该是哪个回调函数。

在 loop() 这个事件循环的函数中，采用了stopped全局变量来控制事件循环的停止，当urls_todo消耗完毕之后，会标记stopped为True。

在事件循环里面有一个阻塞调用，selector.select() 。如果事件不发生，那么应用程序就没事件可处理，所以就干脆阻塞在这里等待事件发生。那可以推断，如果只下载一篇网页，一定要connect()之后才能send()继而recv()，那它的效率和阻塞的方式是一样的。因为不在connect()/recv()上阻塞，也得在select()上阻塞。

所以，selector机制(后文以此称呼代指epoll/kqueue)是设计用来解决大量并发连接的。当系统中有大量非阻塞调用，能随时产生事件的时候，selector机制才能发挥最大的威力。

在单线程内用 事件循环+回调 搞定了10篇网页同时下载的问题。这，已经是异步编程了。虽然有一个for 循环顺序地创建Crawler 实例并调用 fetch 方法，但是fetch 内仅有connect()和注册可写事件，而且从执行时间明显可以推断，多个下载任务确实在同时进行！

上述代码异步执行的过程：

• 创建Crawler 实例；

• 调用fetch方法，会创建socket连接和在selector上注册可写事件；

• fetch内并无阻塞操作，该方法立即返回；

• 重复上述3个步骤，将10个不同的下载任务都加入事件循环；

• 启动事件循环，进入第1轮循环，阻塞在事件监听上；

• 当某个下载任务EVENT_WRITE被触发，回调其connected方法，第一轮事件循环结束；

• 进入第2轮事件循环，当某个下载任务有事件触发，执行其回调函数；此时已经不能推测是哪个事件发生，因为有可能是上次connected里的EVENT_READ先被触发，也可能是其他某个任务的EVENT_WRITE被触发；（此时，原来在一个下载任务上会阻塞的那段时间被利用起来执行另一个下载任务了）

• 循环往复，直至所有下载任务被处理完成

• 退出事件循环，结束整个下载程序

做异步编程，上述的“事件循环+回调”这种模式是逃不掉的，尽管它可能用的不是epoll，也可能不是while循环。

但是在某些异步编程中并没有看到 CallBack 模式呢？比如Python的异步编程中，其主角是协程。

 

协程与asyncio

 

协程

协程(Co-routine)，即是协作式的例程

它是非抢占式的多任务子例程的概括，可以允许有多个入口点在例程中确定的位置来控制程序的暂停与恢复执行。

例程是什么？编程语言定义的可被调用的代码段，为了完成某个特定功能而封装在一起的一系列指令。一般的编程语言都用称为函数或方法的代码结构来体现。

首先，要知道的是，无论是多进程，多线程还是协程，都是为了解决多任务同时进行的问题。而多任务系统实现的关键在于如何暂停当前任务，保存当前任务的上下文，选择下一个任务，恢复下一个任务的上下文 ，执行下一个任务。

对计算机的不同层次来说，上下文的含义也不一样。

• 对CPU来说，上下文就是操作数寄存器、栈寄存器、状态寄存器等各类寄存器。
• 对于进程来说，上下文就是，寄存器、信号、分配的内存空间，文件描述符等各类由 cpu 抽象的出来的硬件资源
• 对于线程来说，上下文就是，寄存器，线程堆栈···
• 对于一个函数来说，上下文就是当前的命名空间。

进程的切换 需要 切换系统资源和指令，消耗时间最长。

线程的切换，不需要切换系统资源，只需要切换指令、线程堆栈。但这个过程也需要系统调用。

协程的切换都在用户空间内进行，不需要进行系统调用。

在Python中线程切换，是由 python 虚拟机控制，通过一个系统调用，来进行线程切换。协程的切换过程完全由程序自身控制。

协程优于线程的主要在于

• python 线程调度方式是，Python2.x中每执行 100 个字节码或者遇到阻塞就停止当前线程，在Python3中则是使用固定时间了，而不再是ticks计数达到100释放GIL，然后进行一个系统调用，让 os 内核选出下一个线程。但是协程 只会在 阻塞的时候，切换到下一个协程。100个字节码，说多不多，说少不少，可能调用两个库函数说不定就没了，因此线程的切换存在很多是无效的切换，当线程数量越大，这种因为调度策略的先天不足带来的性能损耗就越大。
• 线程需要进行系统调用，协程不需要。系统调用需要进入内核态，无效的调度会让这部分开销显得更大
• 协程可以自主调度，而线程只能决定合适退出，但是下一个线程是谁则依赖于操作系统。

Python中，协程有两种，一种无栈协程，python 中以 asyncio 为代表，一种有栈协程，python 中以 gevent 为代表。

 

yield 和 yield from

 

yield

目前最新的Python已经没有采用基于 yield 的协程了。但是为了更好的理解协程，先来一个简单的 yield 的小例子。

def fun_e():

    print('yield 1')

    yield 1

    print('yield 2')

    yield 2



gen = fun_e()

print('start')

a = gen.send(None)

print('生成器的第一个值', a)

b = gen.next(None)

print('生成器的第二个值', b)

复制代码

 

上面代码的输出结果是

start

yield 1

生成器的第一个值 1

yield 2

生成器的第二个值 2

复制代码

 

这里程序运行到第一个 yield 的时候，保存了函数的上下文之后便退出了，然后又通过 next 方法进入了这个函数，将刚刚保存的函数上下文恢复并继续运行。

一个协程程序的所有就是：保存上下文 切换运行程序 恢复上下文 重新进入程序。

Cpython中的上下文，被封装成了一个PyFrameObject的结构，也可以叫它栈帧。

源码：

typedef struct _frame {

    PyObject_VAR_HEAD

    struct _frame *f_back;      /* previous frame, or NULL */

    PyCodeObject *f_code;       /* code segment */

    PyObject *f_builtins;       /* builtin symbol table (PyDictObject) */

    PyObject *f_globals;        /* global symbol table (PyDictObject) */

    PyObject *f_locals;         /* local symbol table (any mapping) */

    PyObject **f_valuestack;    /* points after the last local */

    /* Next free slot in f_valuestack.  Frame creation sets to f_valuestack.

       Frame evaluation usually NULLs it, but a frame that yields sets it

       to the current stack top. */

    PyObject **f_stacktop;

    PyObject *f_trace;          /* Trace function */

    char f_trace_lines;         /* Emit per-line trace events? */

    char f_trace_opcodes;       /* Emit per-opcode trace events? */



    /* Borrowed reference to a generator, or NULL */

    /* 生成器的指针 */

    PyObject *f_gen;



    int f_lasti;                /* 上一个运行的字节码位置 */

    /* Call PyFrame_GetLineNumber() instead of reading this field

       directly.  As of 2.3 f_lineno is only valid when tracing is

       active (i.e. when f_trace is set).  At other times we use

       PyCode_Addr2Line to calculate the line from the current

       bytecode index. */

    int f_lineno;               /* 对应的Python源码行数 */

    int f_iblock;               /* index in f_blockstack */

    char f_executing;           /* whether the frame is still executing */

    PyTryBlock f_blockstack[CO_MAXBLOCKS]; /* for try and loop blocks */

    PyObject *f_localsplus[1];  /* locals+stack, dynamically sized */

} PyFrameObject;

复制代码

 

在Python的实际执行中，会产生很多的PyFrameObject对象，然后这些对象都被链接起来，形成一条链表。

在Python中的生成器的结构体定义是一个宏，它指向一个PyFrameObject对象，表示这个生成器的上下文。

在生成器这个结构体中，有3个重要的东西：

1. 指向生成器上下文的指针
2. 一个指示生成器状态的字符串 未启动，停止，运行，结束
3. 生成器的字节码

即 上下文 + 指令序列 + 状态

在生成器中，next 和 send 的作用相同，但是 send 可以传入一个参数。

 

yield from

在生成器中，可以使用return返回值，但如果 send 走到 return 语句的时候会报一个StopIteration。 return 返回值的 就在 exception 的 value 中。

如下例：

def test_fun1():

    yield 1

    return 2



gen = test_fun1()

try:

    gen.send(None)

    gen.send(None)

except StopIteration as e:

    print(e.value)

复制代码

 

执行以上代码的输出结果是 2

yield from 有两重性质，一方面，它是一个表达式，表达式自然是有值的，他的值，就是yield from 后面生成器 return 的返回值。非常关键的一点，生成器的 yield 语句会向外产出值，但是 return 的值并不会向外产出。想要获得 return 的返回值，要么用 try 语句捕获异常要么用 yield from 表达式获取值。

可以看一下下面这例子

def test_fun1():

    yield 1

    return 2



def test_fun2():

    a = yield from test_fun1()

    print(f"yield from 表达式的值为 {a}")

    yield None



gen = test_fun2()

gen.send(None)

gen.send(None)

复制代码

 

输出结果：

yield from 表达式的值为 2

yield from 还有一个特点就是可以将内层的生成器的返回值，传到外层。

就像下面这个例子：

def test_gen1():

    yield 1

    yield 2

    return 3



def test_gen2():

    a = yield from test_gen1()

    print(f"yield from {a}")



for i in test_gen2():

    print(i)

复制代码

 

输出结果为：

1

2

yield from 3

复制代码

 

内层生成器 test_gen1() 可以通过 yield from 在最外层将值取出来。

这样我们使用 yield from 可以将多个生成器连接起来。

 

简单理解 yield

一开始接触 yield 的时候很不好理解这个 yield的用法，不明白什么叫做生成器，什么参数传递。其实可以直接把 yield 先简单看成 return，程序执行到 yield 的时候就停止了。

先看一个简单的例子

def example():

    print("开始...")

    while True:

        res = yield 4

        print("res:",res)

g = example()

print(next(g))

print("*"*20)

print(next(g))

复制代码

 

输出结果:

开始...

4

********************

res: None

4

复制代码

 

上述代码的执行顺序为：

1. 程序开始执行以后，因为foo函数中有yield关键字，所以foo函数并不会真的执行，而是先得到一个生成器g(相当于一个对象)
2. 直到调用next方法，foo函数正式开始执行，先执行foo函数中的print方法，然后进入while循环
3. 程序遇到yield关键字，然后把yield想想成return,return了一个4之后，程序停止，并没有执行赋值给res操作，此时next(g)语句执行完成，所以输出的前两行（第一个是while上面的print的结果,第二个是return出的结果）是执行print(next(g))的结果，
4. 程序执行print("*"20)，输出20个
5. 又开始执行下面的print(next(g)),这个时候和上面那个差不多，不过不同的是，这个时候是从刚才那个next程序停止的地方开始执行的，也就是要执行res的赋值操作，这时候要注意，这个时候赋值操作的右边是没有值的（因为刚才那个是return出去了，并没有给赋值操作的左边传参数），所以这个时候res赋值是None,所以接着下面的输出就是res:None,
6. 程序会继续在while里执行，又一次碰到yield,这个时候同样return 出4，然后程序停止，print函数输出的4就是这次return出的4

yield和return的关系和区别了，带yield的函数是一个生成器，而不是一个函数了，这个生成器有一个函数就是next函数，next就相当于“下一步”生成哪个数，这一次的next开始的地方是接着上一次的next停止的地方执行的，所以调用next的时候，生成器并不会从 example 函数的开始执行，只是接着上一步停止的地方开始，然后遇到yield后，return出要生成的数，此步就结束。

再来一个 send 的例子：

def foo():

    print("starting...")

    while True:

        res = yield 4

        print("res:",res)

g = foo()

print(next(g))

print("*"*20)

print(g.send(7))

复制代码

 

输出结果

starting...

4

********************

res: 7

4

复制代码

 

先大致说一下send函数的概念：此时你应该注意到上面那个的紫色的字，还有上面那个res的值为什么是None，这个变成了7，到底为什么，这是因为，send是发送一个参数给res的，因为上面讲到，return的时候，并没有把4赋值给res，下次执行的时候只好继续执行赋值操作，只好赋值为None了，而如果用send的话，开始执行的时候，先接着上一次（return 4之后）执行，先把7赋值给了res,然后执行next的作用，遇见下一回的yield，return出结果后结束。

接上之前的步骤：

1. 程序执行g.send(7)，程序会从yield关键字那一行继续向下运行，send会把7这个值赋值给res变量
2. 由于send方法中包含next()方法，所以程序会继续向下运行执行print方法，然后再次进入while循环
3. 程序执行再次遇到yield关键字，yield会返回后面的值后，程序再次暂停，直到再次调用next方法或send方法。

 

基于生成器的协程

前面说了这么多，在Python里面为什么要使用协程来解决异步的问题呢。我们先来看看前面讲到的“事件循环+回调”这种方式的问题。

 

“事件循环+回调”的问题

在单线程内使用前面爬虫例子中的异步编程，也确实能够大大提高程序运行效率。但是在生产项目中，要应对的复杂度会大大增加。考虑如下问题：

• 如果回调函数执行不正常该如何？
• 如果回调里面还要嵌套回调怎么办？要嵌套很多层怎么办？
• 如果嵌套了多层，其中某个环节出错了会造成什么后果？
• 如果有个数据需要被每个回调都处理怎么办？
• ......

在实际编程中，上述问题不好避免，也确实存在这么些的缺陷。

• 回调层次过多时代码可读性差

    def callback_1():
  
    # processing ...
  
        def callback_2():
  
            # processing.....
  
            def callback_3():
  
                # processing ....
  
                def callback_4():
  
                    #processing .....
  
                    def callback_5():
  
                        # processing ......
  
                    async_function(callback_5)
  
                async_function(callback_4)
  
            async_function(callback_3)
  
        async_function(callback_2)
  
    async_function(callback_1)
  
  复制代码

• 破坏代码结构

  在写同步的程序的时候，代码一般是自上而下运行的。

  fun_1()
  
  fun_2(params)
  
  复制代码

  在上面的代码中，如果 fun_2 函数的处理依赖 fun_1 函数处理的结果，但是 fun_1 函数是异步调用的，那么就不知道 fun_1 什么时候返回值，需要将后续处理的结果以 callback 的形式返回给 fun_1，让 fun_1 执行完之后去执行 fun_2，那么代码变成如下：

  fun_1(fun_2())
  
  复制代码

  如果整个流程都是异步处理，然后流程比较长的话：

  fun_1(fun_2(fun_3(fun_4(......))))
  
  复制代码

如果是同步执行的程序，程序中的每一步都是线程的指令指针控制着的流程，而在回调版本中，流程就是编程的人需要注意和安排的。

• 共享状态管理困难 回顾之前的爬虫代码，同步阻塞版的sock对象从头使用到尾，而在回调的版本中，我们必须在Crawler实例化后的对象self里保存它自己的sock对象。如果不是采用OOP的编程风格，那需要把要共享的状态接力似的传递给每一个回调。多个异步调用之间，到底要共享哪些状态，事先就得考虑清楚，精心设计。

• 错误处理困难 一连串的回调构成一个完整的调用链。例如上述的 a 到 f。假如 d 抛了异常怎么办？整个调用链断掉，接力传递的状态也会丢失，这种现象称为调用栈撕裂。 c 不知道该干嘛，继续异常，然后是 b 异常，接着 a 异常。好嘛，报错日志就告诉你，a 调用出错了，但实际是 d 出错。所以，为了防止栈撕裂，异常必须以数据的形式返回，而不是直接抛出异常，然后每个回调中需要检查上次调用的返回值，以防错误吞没。

代码风格难看是小事，但栈撕裂和状态管理困难这两个缺点会让基于回调的异步编程很艰难。

Python在事件循环+回调的基础上衍生出了基于协程的解决方案，代表作有 Tornado、Twisted、asyncio 等。

 

未来对象

Python 中有种特殊的对象——生成器（Generator），它的特点和协程很像。每一次迭代之间，会暂停执行，继续下一次迭代的时候还不会丢失先前的状态。

为了支持用生成器做简单的协程，Python 2.5 对生成器进行了增强(PEP 342)，该增强提案的标题是 “Coroutines via Enhanced Generators”。有了PEP 342的加持，生成器可以通过yield 暂停执行和向外返回数据，也可以通过send()向生成器内发送数据，还可以通过throw()向生成器内抛出异常以便随时终止生成器的运行。

这里我们不用回调的方式了，怎么知道异步调用的结果呢？先设计一个对象，异步调用执行完的时候，就把结果放在它里面。这种对象称之为未来对象。

class Future:

    """

    未来对象

    异步调用执行完的时候，就把结果放在它里面。

    """

    def __init__(self):

        self.result = None

        self._callbacks = []



    def add_done_callback(self, fn):

        self._callbacks.append(fn)



    def set_result(self, result):

        self.result = result

        for fn in self._callbacks:

            fn(self)

复制代码

 

未来对象有一个result属性，用于存放未来的执行结果。还有个set_result()方法，是用于设置result的，并且会在给result绑定值以后运行事先给future添加的回调。回调是通过未来对象的add_done_callback()方法添加的。

虽然这个地方还是有 callback，但是这个 callback 和之前的不太一样。

 

重构Crawler爬虫

因为有了未来对象，我们先用Future来重构一下爬虫。

class Crawler:

    def __init__(self, url):

        self.url = url

        self.response = b''



    def fetch(self):

        sock = socket.socket()

        sock.setblocking(False)

        try:

            sock.connect(('example.com', 80))

        except BlockingIOError:

            pass

        f = Future()



        def on_connect():

            f.set_result(None)



        selector.register(sock.fileno(), EVENT_WRITE, on_connect)

        yield f

        selector.unregister(sock.fileno())

        get = 'GET {0} HTTP/1.0 \r\nHost: example.com\r\n\r\n'.format(self.url)

        sock.send(get.encode('ascii'))



        global stopped

        while True:

            f = Future()



            def on_readable():

                f.set_result(sock.recv(4096))



            selector.register(sock.fileno(), EVENT_READ, on_readable)

            chunk = yield f

            selector.unregister(sock.fileno())

            if chunk:

                self.response += chunk

            else:

                urls_todo.remove(self.url)

                if not urls_todo:

                    stopped = True

                    break

复制代码

 

这里的 fetch方法内有了yield表达式，所以它成为生成器。生成器需要先调用next()迭代一次或者是先send(None)启动，遇到yield之后便暂停。那这fetch生成器如何再次恢复执行呢？但是目前还有生成器的启动代码。这里我们需要添加一个任务对象(Task)来启动它。

 

任务对象(Task)

遵循一个编程规则：单一职责，每种角色各司其职。目前还没有一个角色来负责生成器的执行和管理生成器的状态，那么我们就创建一个。

class Task:

    """任务对象"""

    def __init__(self, coro):

        self.coro = coro

        f = Future()

        f.set_result(None)

        self.step(f)

    

    def step(self, future):

        try:

            # send放到coro执行，即fetch，直到下次yield

            # next_future为yield返回对象

            next_future = self.coro.send(future.result)

        except StopIteration:

            return 

        next_future.add_done_callback(self.step)

复制代码

 

上述代码中Task封装了coro对象，即初始化时传递给他的对象，被管理的任务是待执行的协程，故而这里的coro就是fetch()生成器。它还有个step()方法，在初始化的时候就会执行一遍。step()内会调用生成器的send()方法，初始化第一次发送的是None就驱动了coro即fetch()的第一次执行。

send()完成之后，得到下一次的future，然后给下一次的future添加step()回调。add_done_callback()其实不是给写爬虫业务逻辑用的。

再看一下fetch()生成器，其内部写完了所有的业务逻辑，包括如何发送请求，如何读取响应。而且注册给 selector 的回调相当简单，就是给对应的 future 对象绑定结果值。两个 yield 表达式都是返回对应的 future 对象，然后返回 Task.step() 之内，这样 Task, Future, Coroutine三者就串联在了一起。

初始化Task对象以后，把fetch()给驱动到了第 yied f 就完事了，接下来应该怎么继续。

 

事件循环(Event Loop)驱动协程运行

接下来，只需等待已经注册的EVENT_WRITE事件发生。事件循环就像心脏一般，只要它开始跳动，整个程序就会持续运行。

def loop():

    """事件循环驱动协程"""

    while not stopped:

        events = selector.select()

        for event_key, event_mask in events:

            callback = event_key.data

            callback()

复制代码

   

整个重构之后的爬虫

完整的程序如下：

import socket

from selectors import EVENT_READ, EVENT_WRITE, DefaultSelector

from socket import create_connection



selector = DefaultSelector()

stopped = False

urls_todo = {'/', '/1', '/2', '/3', '/4', '/5', '/6', '/7', '/8', '/9'}





class Future:

    """

    未来对象

    异步调用执行完的时候，就把结果放在它里面。

    """

    def __init__(self):

        self.result = None

        self._callbacks = []



    def add_done_callback(self, fn):

        self._callbacks.append(fn)



    def set_result(self, result):

        self.result = result

        for fn in self._callbacks:

            fn(self)





class Crawler:

    def __init__(self, url):

        self.url = url

        self.response = b''



    def fetch(self):

        sock = socket.socket()

        sock.setblocking(False)

        try:

            sock.connect(('example.com', 80))

        except BlockingIOError:

            pass

        f = Future()



        def on_connect():

            f.set_result(None)



        selector.register(sock.fileno(), EVENT_WRITE, on_connect)

        yield f

        selector.unregister(sock.fileno())

        get = 'GET {0} HTTP/1.0 \r\nHost: example.com\r\n\r\n'.format(self.url)

        sock.send(get.encode('ascii'))



        global stopped

        while True:

            f = Future()



            def on_readable():

                f.set_result(sock.recv(4096))



            selector.register(sock.fileno(), EVENT_READ, on_readable)

            chunk = yield f

            selector.unregister(sock.fileno())

            if chunk:

                self.response += chunk

            else:

                urls_todo.remove(self.url)

                if not urls_todo:

                    stopped = True

                    break





class Task:

    """任务对象"""

    def __init__(self, coro):

        self.coro = coro

        f = Future()

        f.set_result(None)

        self.step(f)

    

    def step(self, future):

        try:

            # send放到coro执行，即fetch，直到下次yield

            # next_future为yield返回对象

            next_future = self.coro.send(future.result)

        except StopIteration:

            return 

        next_future.add_done_callback(self.step)



def loop():

    """事件循环驱动协程"""

    while not stopped:

        events = selector.select()

        for event_key, event_mask in events:

            callback = event_key.data

            callback()



if __name__ == "__main__":

    import time

    start = time.time()

    for url in urls_todo:

        crawler = Crawler(url)

        Task(crawler.fetch())

    loop()

    print(time.time() - start)



"""

现在loop有了些许变化，callback()不再传递event_key和event_mask参数。也就是说，

这里的回调根本不关心是谁触发了这个事件,

结合fetch()可以知道，它只需完成对future设置结果值即可f.set_result()。

"""

复制代码

   

生成器风格和回调风格总结

在回调风格中：

• 存在链式回调（虽然示例中嵌套回调只有一层）

• 请求和响应也不得不分为两个回调以至于破坏了同步代码那种结构

• 程序员必须在回调之间维护必须的状态。

而基于生成器协程的风格：

• 无链式调用

• selector的回调里只管给future设置值，不再关心业务逻辑

• loop 内回调callback()不再关注是谁触发了事件

• 已趋近于同步代码的结构

• 无需程序员在多个协程之间维护状态，例如哪个才是自己的sock

 

yield from 改进生成器协程

如果说fetch的容错能力要更强，业务功能也需要更完善，怎么办？而且技术处理的部分（socket相关的）和业务处理的部分（请求与返回数据的处理）混在一起。

• 创建socket连接可以抽象复用吧？

• 循环读取整个response可以抽象复用吧？

• 循环内处理socket.recv()的可以抽象复用吧？

但是这些关键节点的地方都有yield，抽离出来的代码也需要是生成器。而且fetch()自己也得是生成器。生成器里捣鼓生成器，好像有些麻烦。

好在有 yield from 来解决这个问题。

 

yield from 语法

yield from 是Python 3.3 新引入的语法（PEP 380）。它主要解决的就是在生成器里弄生成器不方便的问题。它有两大主要功能。

第一个功能是：让嵌套生成器不必通过循环迭代yield，而是直接yield from。以下两种方式是等价的。

def gen_one():

    subgen = range(10)    

    yield from subgen



def gen_two():

    subgen = range(10)    

    for item in subgen:        

        yield item

复制代码

 

第二个功能就是在子生成器和原生成器的调用者之间打开双向通道，两者可以直接通信。

def gen():

    yield from subgen()def subgen():

    while True:

        x = yield

        yield x+1def main():

    g = gen()



    next(g)                # 驱动生成器g开始执行到第一个 yield

    retval = g.send(1)     # 看似向生成器 gen() 发送数据

    print(retval)          # 返回2

    g.throw(StopIteration) # 看似向gen()抛入异常

复制代码

 

通过上述代码清晰地理解了yield from的双向通道功能。关键字yield from在gen()内部为subgen()和main()开辟了通信通道。main()里可以直接将数据1发送给subgen(),subgen()也可以将计算后的数据2返回到main()里，main()里也可以直接向subgen()抛入异常以终止subgen()。

 

重构代码

首先我们需要将 Future 对象变成一个 iter 对象：

class Future:

    """

    未来对象

    异步调用执行完的时候，就把结果放在它里面。

    """

    def __init__(self):

        self.result = None

        self._callbacks = []



    def add_done_callback(self, fn):

        self._callbacks.append(fn)



    def set_result(self, result):

        self.result = result

        for fn in self._callbacks:

            fn(self)



    def __iter__(self):

        # 将Future变成一个iter对象

        yield self

        return self.result

复制代码

 

之后

抽象socket连接的功能：

def connect(sock, address):

    f = Future()

    sock.setblocking(False)

    try:

        sock.connect(address)

    except BlockingIOError:

        pass



    def on_connected():

        f.set_result(None)



    selector.register(sock.fileno(), EVENT_WRITE, on_connected)

    yield from f

    selector.unregister(sock.fileno())

复制代码

 

抽象单次recv()和读取完整的response功能

def read(sock):

    f = Future()



    def on_readable():

        f.set_result(sock.recv(4096))



    selector.register(sock.fileno(), EVENT_READ, on_readable)

    chunk = yield from f

    selector.unregister(sock.fileno())

    return chunk



def read_all(sock):

    response = []

    chunk = yield from read(sock)

    while chunk:

        response.append(chunk)

        chunk = yield from read(sock)

    return b''.join(response)

复制代码

 

现在重构Crawler类

class Crawler:

    

    def __init__(self, url):

        self.url = url

        self.response = b''



    def fetch(self):

        global stopped

        sock = socket.socket()

        yield from connect(sock, ('example.com', 80))

        get = f'GET {self.url} HTTP/1.0\r\nHost: example.com\r\n\r\n'

        sock.send(get.encode('ascii'))

        self.response = yield from read_all(sock)

        urls_todo.remove(self.url)

        if not urls_todo:

            stopped = True

复制代码

 

在Python 3.3 引入yield from新语法之后，就不再推荐用yield去做协程。全都使用yield from由于其双向通道的功能，可以让我们在协程间随心所欲地传递数据。

有了 yield from 这个工具，我们便可以将多个生成器串联起来。yield from 的意义在于，将这些生成器串联起来形成一颗树，并且提供了一种便捷的方法，将这颗树的叶子节点依次返回。yield from 将多个生成器连接起来的方式，我们可以使用很简单的方式就可以将所有的 yield 返回值一一提取出来。不断的对根节点的生成器 进行send 操作即可。

上面的例子介绍了 yield 和 yield from，但是它们和 asyncio 之间有什么区别和联系呢，来看一下 asyncio。

 

asyncio

 

asyncio 简介

asyncio是Python 3.4 试验性引入的异步I/O框架（PEP 3156），提供了基于协程做异步I/O编写单线程并发代码的基础设施。其核心组件有事件循环（Event Loop）、协程(Coroutine）、任务(Task)、未来对象(Future)以及其他一些扩充和辅助性质的模块。

在引入asyncio的时候，还提供了一个装饰器@asyncio.coroutine用于装饰使用了yield from的函数，以标记其为协程。但并不强制使用这个装饰器。

虽然发展到 Python 3.4 时有了yield from的加持让协程更容易了，但是由于协程在Python中发展的历史包袱所致，很多人仍然弄不明白生成器和协程的联系与区别，也弄不明白yield 和 yield from 的区别。这种混乱的状态也违背Python之禅的一些准则。

于是Python设计者们又快马加鞭地在 3.5 中新增了async/await语法（PEP 492），对协程有了明确而显式的支持，称之为原生协程。async/await 和 yield from这两种风格的协程底层复用共同的实现，而且相互兼容。

在Python 3.6 中asyncio库“转正”，不再是实验性质的，成为标准库的正式一员。

 

asyncio重写爬虫

用 asyncio 重写一下之前写的例子：

import asyncio

import aiohttp



host = 'http://example.com'

urls_todo = {'/', '/1', '/2', '/3', '/4', '/5', '/6', '/7', '/8', '/9'}





async def fetch(url):

    async with aiohttp.ClientSession(loop=loop) as session:

        async with session.get(url) as response:

            response = await response.read()

            return response



if __name__ == '__main__':

    import time

    start = time.time()

    loop = asyncio.get_event_loop()

    tasks = [fetch(host + url) for url in urls_todo]

    loop.run_until_complete(asyncio.gather(*tasks))

    print(time.time() - start)

复制代码

 

上述代码运行时间大概为0.36s

对比起之前的，变化很大：

• 没有了yield 或 yield from，而是async/await
• 没有了自造的loop()，取而代之的是asyncio.get_event_loop()
• 无需自己在socket上做异步操作，不用显式地注册和注销事件，aiohttp库已经代劳
• 没有了显式的 Future 和 Task，asyncio已封装
• 更少量的代码，更优雅的设计

和同步阻塞版的代码对比：

• 异步化
• 代码量相当（引入aiohttp框架后更少）
• 代码逻辑同样简单，跟同步代码一样的结构、一样的逻辑
• 接近10倍的性能提升

 

简单实现 asyncio

 

asyncio逻辑梳理

我们将之前的实现异步的代码抽取出来，主要就有 Future, Task, event_loop这三个东西。

from selectors import EVENT_READ, EVENT_WRITE, DefaultSelector





stopped = False



class Future:



    def __init__(self):

        self.result = None

        self._callbacks = []



    def add_done_callback(self, fn):

        self._callbacks.append(fn)



    def set_result(self, result):

        self.result = result

        for fn in self._callbacks:

            fn(self)



    def __iter__(self):

        yield self

        return self.result





class Task:

    

    def __init__(self, coro):

        self.coro = coro

        f = Future()

        f.set_result(None)

        self.step(f)  #激活 Task 包裹的生成器



    def step(self, future):

        try:

            next_future = self.coro.send(future.result)

        except StopIteration:

            return

        next_future.add_done_callback(self.step)



selector = DefaultSelector()



def loop():

    while not stopped:

        events = selector.select()

        for event_key, event_mask in events:

            callback = event_key.data

            callback()

复制代码

 

这个代码当然是不能直接使用的，我们需要改造一下。

asyncio库里面的核心也是这三个东西，Future, Task 还有事件循环。

在开始之前，我们先使用 asyncio来写个小demo，根据这个demo来梳理一下asyncio的逻辑。

import asyncio



async def get_html(url):

    print("开始获取HTML")

    await asyncio.sleep(2)

    print("结束获取HTML")



if __name__ == "__main__":

    loop = asyncio.get_event_loop()

    func = get_html("https://www.baidu.com")

    task = loop.create_task(func)

    loop.run_until_complete(task)

复制代码

 

在上述代码中，我们构建了一个异步函数 get_html 来模拟访问网页的过程。

在这个代码段中，首先执行的就是 get_event_loop 这个函数，这个函数的作用就是获取一个事件循环，不停的循环检测是否有事件准备好，如果检查到准备好的，就调用注册在事件上的回调函数，直到 stopped 置位时退出循环。 get_event_loop() 会返回一个事件循环类的实例，这个类继承于 BaseEventLoop。

之后执行了 create_task() 这个函数，来看看它都干嘛了。

在asyncio/base_events.py中找到BaseEventLoop定义，在BaseEventLoop类中有个create_task方法。

这个方法的核心代码就只有这俩。

def create_task(self, coro):    

        task = tasks.Task(coro, loop=self)

        return task

复制代码

 

就是创建了一个Task的实例，然后将传入的协程 coro 通过 Task 这个类中的逻辑进行一步一步驱动。

每个 Task 类实例都会包裹一个协程(coro)，然后通过函数 step 中的 send 来驱动协程。这个协程经过 future 一步一步驱动起来。

在 Task 类中，初始化实例的时候将协程保存，然后通过 step 函数来启动协程。但是这里的 step 函数和上面我们抽取出来的不太一样，它多了很多参数检查和异常处理，然后就是，我们上面抽取出来的 Task 类在 init 的时候，就将整个协程驱动起来，但是在 asyncio 中，它不会马上调用 step 函数，而是在下一帧(下一次循环)中调用(_loop.call_soon 函数)。

这里调用 call_soon 函数就是将 Task 实例的 _step 函数添加到待执行的队列中去，这个函数也是定义在 asyncio/base_events.py 的 BaseEventLoop 类中的。

    def _call_soon(self, callback, args):

        handle = events.Handle(callback, args, self)

        if handle._source_traceback:

            del handle._source_traceback[-1]

        self._ready.append(handle)  # 事件添加到队列

        return handle

复制代码

 

它返回了一个 Handle 类的实例。这里的 Handle 类就是包裹了就绪事件的回调函数的，其中定义了一个run方法，就是直接执行回调函数，self._ready 保存着 Handle 类的实例，在 asyncio 中 loop 死循环不断检测是否有事件就绪，即检测 self._ready是否有为空，不为空就从其中弹出 Handle 实例，然后调用handle实例的run方法，其实就是执行注册在就绪事件上的回调函数。一旦有就绪事件，就调用其回调函数。

现在在我们写的那个小 demo 中，已经通过 task = loop.create_task(func) 创建了一个 task 实例，该 task 实例包裹了我们自己定义的协程 func ，并且 在task 初始化的时候在 __init__ 函数中通过 call_soon 通知下一次循环立即执行 task 的_step函数来激活cora协程。接下来就是run_until_complete函数了。

run_until_complete函数同样定义在asyncio/base_events.py的BaseEventLoop类中。这个函数中就有 loop 的死循环。(节选的代码，删除了部分代码)

def run_until_complete(self, future):

        future = tasks.ensure_future(future, loop=self)  # ensure_future，即，确保是future。返回的是future（task也是future）



        future.add_done_callback(_run_until_complete_cb)  # 用来结束循环

        try:

            self.run_forever()

        except:

            if new_task and future.done() and not future.cancelled():

                future.exception()

            raise

        finally:

            future.remove_done_callback(_run_until_complete_cb)

        if not future.done():

            raise RuntimeError('Event loop stopped before Future completed.')



        return future.result()

复制代码

 

函数首先确保传递进来的参数是future，Task 是继承 Future的，所以 task 也是 future。我们外面传进来的参数是个task实例，所以这个函数调用返回的其实就是本身（传进去是啥返回就是啥），然后给我们传进来的task实例通过调用add_done_callback添加_run_until_complete_cb回调函数，这个回调函数比较关键，run_until_complete的做的最重要的事就是给传进来的task实例添加这个回调，点进_run_until_complete_cb，可以看到就是调用了loop的stop函数，这个的意义就是，当我们传进来的task包裹的协程运行结束后，就调用这个回调，跳出循环（就是相当于我们抽取出来的代码中的stopped变量的作用），否则死循环就真的是死循环了，永远跳不出。

之后就是真的死循环，run forever。

关键代码

def run_forever(self):try:

            events._set_running_loop(self)

            while True:

                self._run_once()

                if self._stopping:

                    break

        finally:

            ...

复制代码

 

这个函数不断的调用_run_once()，就像我们抽取出来的loop函数中不断地调用下面这段代码：

events = selector.select()

        for event_key, event_mask in events:

            callback = event_key.data

            callback()

复制代码

 

而在 _run_once()中：

        else:

            event_list = self._selector.select(timeout)  # 筛选就绪事件，将其回调添加到self._ready中

        self._process_events(event_list)  # 该函数具体实现在selector_events.py中

复制代码

 

这里也就是选出就绪事件，然后添加到self._ready队列中，随后执行。在_run_once()的尾部，我们看到如下代码：

ntodo = len(self._ready)

        for i in range(ntodo):

            handle = self._ready.popleft()

            if handle._cancelled:

                continue

            if self._debug:

                try:

                    self._current_handle = handle

                    t0 = self.time()

                    handle._run()

                    dt = self.time() - t0

                    if dt >= self.slow_callback_duration:

                        logger.warning('Executing %s took %.3f seconds',

                                       _format_handle(handle), dt)

                finally:

                    self._current_handle = None

            else:

                handle._run()

        handle = None  # Needed to break cycles when an exception occurs.

复制代码

 

这里就是调用就绪事件的回调函数的执行。先看_ready队列中是否有待处理的Handle实例，如果有，那就一个一个执行，handle中的_run()方法就是执行就绪事件的回调函数。至此，就把我们抽取出来的中的loop()函数的逻辑对应到了asyncio源码的循环之中。

最后来看看Future

正如我们上面抽取的代码中的Future：

    def __iter__(self):

        yield self

        return self.result  # 在Task.step中send(result)的时候再次调用这个生成器，但是此时会抛出stopInteration异常，并且把self.result返回

复制代码

 

yield的出现使得__iter__函数变成一个生成器，生成器本身就有next方法，所以不需要额外实现。yield from x语句首先调用iter(x)获取一个迭代器（生成器也是迭代器）。

这里的future和asyncio中的future，结构是一样的，功能也类似。最后运行起来的时候就是生成器一层嵌套一层。

 

总结

以上介绍了Python异步编程和自己简单实现一个asyncio。如果还有不太理解，或者是觉得文章在某些地方还有需要提升的地方，以及有不同观点的地方，欢迎在下面留言交流，帮助我一起把这篇文章变得更好。

 

最后