序言
 第1章并行和分布式计算介绍
第2章异步编程
第3章 Python的并行计算
 第4章 Celery分布式应用
 第5章云平台部署Python
第6章超级计算机群使用Python
第7章测试和调试分布式应用
 第8章继续学习

从本章开始，终于开始写代码了！本书中所有的代码都适用于Python 3.5及以上版本。当模块、语句或语法结构不适用于以前的版本时（比如Python 2.7），会在本章中指出。进行一些修改，本书代码也可以运行在Python 2.x版本上。

先回顾下上一章的知识。我们已经学到，改变算法的结构可以让其运行在本地计算机，或运行在集群上。即使是在一台计算机上运行，我们也可以使用多线程或多进程，让子程序运行在多个CPU上。

现在暂时不考虑多CPU，先看一下单线程/进程。与传统的同步编程相比，异步编程或非阻塞编程，可以使性能获得极大提高。

任何包含多任务的程序，它的每个每个任务都在执行一个操作。我们可以把这些任务当做功能或方法，也可以把几个任务合并看做一个功能。例如，将总任务细分、在屏幕打印内容、或从网络抓取信息，等等。

看一下传统程序中的这些任务是如何使用一个CPU的。考虑一个原生的实例，它有四个任务：A、B、C、D。这些任务具体是做什么在这里不重要。我们可以假设这四个任务是关于计算和I/O操作的。安排这四个任务的最直观的方式是序列化。下图展示了这四个任务对CPU的使用：

我们看到，当每个任务都执行I/O操作时，CPU处于空闲状态，等待任务进行计算。这使得CPU大部分时间处于闲置状态。

重点是，从不同组件，例如硬盘、内存和网络，向CPU传递数据的速度相差极大（几个数量级）。

这就会造成任何进行大量I/O操作（访问硬盘、网络通讯等等）的代码都极有可能造成CPU大部分时间闲置，如上图所示。

理想的状态应该是安排一下任务，当一个任务等待I/O时，它处于悬停状态，就让另一个任务接管CPU。这就是异步（也称为事件驱动）编程。

下图生动地展示了用异步编程的方式安排四个任务：

任务仍然是序列的，但是不再各自占用CPU直到任务结束，任务不需要计算时，它们会自发地放弃CPU。尽管CPU仍有闲置，程序的总运行时间明显缩短了。

使用多线程在不同的线程并行运行，也可以达到同样的效果。但是，有一个显著的不同：使用多线程时，是由操作系统决定哪个线程处于运行或悬停。然而，在异步编程中，每个任务可以自己决定是否放弃CPU。

另外，单单使用异步编程，我们不能做出真正的并发：同一时间仅仅有一个任务在运行，消除了竞争条件。当然，我们可以混合使用多线程/多进程和异步编程。

另一点要注意的是，异步编程更善于处理I/O密集型任务，而不是CPU密集型任务（暂停任务不会使性能提高）。

协程

在Python中，让一个功能中途暂停的关键是使用协程。为了理解协程，先要理解生成器generator，要理解生成器，先要理解迭代器iterator。

大部分Python开发者都熟悉对类进行迭代（例如，字符串、列表、元组、文件对象等等）：

>>> for i in range(3):
...     print(i)
...
1
2
>>> for line in open('exchange_rates_v1.py'):
...     print(line, end='')
... 
#!/usr/bin/env python3
import itertools
import time
import urllib.request
…

我们可以将各种对象（不仅仅是列表和字符串）进行迭代的原因是迭代协议。迭代协议定义了迭代的标准格式：一个执行__iter__和__next__（或Python 2.x中的 __iter__和next）的对象就是一个迭代器，可以进行迭代操作，如下所示：

class MyIterator(object):
    def __init__(self, xs):
        self.xs = xs

    def __iter__(self):
        return self

    def __next__(self):
        if self.xs:
            return self.xs.pop(0)
        else:
            raise StopIteration

for i in MyIterator([0, 1, 2]):
    print(i)

结果如下所示：

1
2
3

我们能对MyIterator中的实例进行循环的原因是，它用__iter__和__next__方法，运行了迭代协议：前者返回了迭代的对象，后者逐个返回了序列中的元素。

为了进一步理解协议是如何工作的，我们手动分解这个循环，如下所示：

itrtr = MyIterator([3, 4, 5, 6])

print(next(itrtr))
print(next(itrtr))
print(next(itrtr))
print(next(itrtr))

print(next(itrtr))

运行结果如下：

3
4
5
6
Traceback (most recent call last):
  File "iteration.py", line 32, in <module>
    print(next(itrtr))
  File "iteration.py", line 19, in __next__
  raise StopIteration
StopIteration

我们实例化了MyIterator，然后为了获取它的值，我们多次调用了next()。当序列到头时，next()会抛出异常StopIteration。Python中的for循环使用了同样的机制，它调用迭代器的next()，通过获取异常StopIteration得知何时停止。

生成器就是一个callable，它生成一个结果序列，而不是返回结果。这是通过产生（通过yield关键字）值而不是返回值，见下面的例子（generators.py）：

def mygenerator(n):
    while n:
        n -= 1
        yield n

if __name__ == '__main__':
    for i in mygenerator(3):
        print(i)

结果如下：

2
1
0

这是一个使用yield使mygenerator成为生成器的简单例子，它的功能并不简单。调用generator函数并不开始生成序列，只是产生一个generator对象，见如下shell语句：

>>> from generators import mygenerator
>>> mygenerator(5)
<generator object mygenerator at 0x101267b48>

为了激活generator对象，需要调用next()，见如下代码：

>>> g = mygenerator(2)
>>> next(g)
1
>>> next(g)
0
>>> next(g)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
StopIteration

每个next()从生成的序列产生一个值，直到序列为空，也就是获得异常StopIteration时。迭代器的行为也是类似的。本质上，生成器是简化的迭代器，免去了定义类中__iter__和__next__的方法。

另外，生成器是一次性操作，不能重复生成的序列。若要重复序列，必须再次调用generator函数。

用来在generator函数中产生序列值的yield表达式，还可以在等号右边使用，以消除值。这样就可以得到协程。协程就是一类函数，它可以通过yield，在指定位置暂停或继续任务。

需要注意，尽管协程是强化的生成器，在概念意义上并不等于生成器。原因是，协程与迭代无关。另一不同点，生成器产生值，而协程消除值。

让我们做一些协程，看看如何使用。协程有三种主要的结构，如下所示：

yield()：用来暂停协程的执行
send()：用来向协程传递数据（以让协程继续执行）
close()：用来关闭协程

下面代码展示了协程的使用（coroutines.py）：

def complain_about(substring):
    print('Please talk to me!')
    try:
        while True:
            text = (yield)
            if substring in text:
                print('Oh no: I found a %s again!'
                      % (substring))
    except GeneratorExit:
        print('Ok, ok: I am quitting.')

我们先定义个一个协程，它就是一个函数，名字是complain_about，它有一个参数：一个字符串。打印一句话之后，进入一个无限循环，由try except控制退出，即只有通过异常才能退出。利用异常GeneratorExit，当获得这个异常时就会退出。

循环的主体十分简单，使用yield来获取数据，存储在变量text中。然后，我们检测substring是否在text中。如果在的话，弹出一条新语句。

下面代码展示了在shell中如何使用这个协程：

>>> from coroutines import complain_about
>>> c = complain_about('Ruby')
>>> next(c)
Please talk to me!
>>> c.send('Test data')
>>> c.send('Some more random text')
>>> c.send('Test data with Ruby somewhere in it')
Oh no: I found a Ruby again!
>>> c.send('Stop complaining about Ruby or else!')
Oh no: I found a Ruby again!
>>> c.close()
Ok, ok: I am quitting.

执行complain_about('Ruby')产生了协程。为了使用新建的协程，我们用next()调用它，与在生成器中所做的相同。只有调用next()之后，才在屏幕上看到Please talk to me!。

这时，协程到达了text = (yield)，意味着它暂停了执行。控制点返回了shell，我们就可以向协程发送数据了。我们使用send()方法发送数据，如下所示：

>>> c.send('Test data')
>>> c.send('Some more random text')
>>> c.send('Test data with Ruby somewhere in it')
Oh no: I found a Ruby again!

每次调用send()方法都使代码到达下一个yield。在我们的例子中，到达while循环的下一次迭代，返回text = (yield)。这里，控制点返回shell。

我们可以调用close()方法停止协程，它可以在协程内部抛出异常GeneratorExit。此时，协程唯一能做的就是清理数据并退出。下面的代码展示了如何结束协程：

>>> c.close()
Ok, ok: I am quitting.

如果将try except部分注释掉，就不会获得GeneratorExit异常。但是协程还是会停止，如下所示：

>>> def complain_about2(substring):
...     print('Please talk to me!')
...     while True:
...         text = (yield)
...         if substring in text:
...             print('Oh no: I found a %s again!'
...                   % (substring))
... 
>>> c = complain_about2('Ruby')
>>> next(c)
Please talk to me!
>>> c.close()
>>> c.send('This will crash')
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
StopIteration
>>> next(c)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
StopIteration

我们看到，一旦关闭协程，对象仍会保持，但是用途为零：不能向它发送数据，也不能调用next()使用它。

当使用协程时，许多人觉得必须要用next()很繁琐，转而使用装饰器，避免多余的调用，如下所示：

>>> def coroutine(fn):
...     def wrapper(*args, **kwargs):
...         c = fn(*args, **kwargs)
...         next(c)
...         return c
...     return wrapper
... 
>>> @coroutine
... def complain_about2(substring):
...     print('Please talk to me!')
...     while True:
...         text = (yield)
...         if substring in text:
...             print('Oh no: I found a %s again!'
...                   % (substring))
... 
>>> c = complain_about2('JavaScript')
Please talk to me!
>>> c.send('Test data with JavaScript somewhere in it')
Oh no: I found a JavaScript again!
>>> c.close()

协程的层级结构可以很复杂，可以让一个协程向其它多个协程发送数据，或从多个源接收数据。这在网络集群编程和系统编程中很有用（为了提高性能），可以用纯Python高效代替大多数Unix工具。

一个异步实例

为了简单又有趣，让我们写一个工具，可以对指定的文件，统计某个词的出现次数。使用之前的协程做基础，再添加一些功能。

在Linux和Mac OS X上，可以使用grep命令获得同样的结果。我们先下载一个大的文本文件，用作输入的数据。我们选择的是Project Gutenberg上列夫托尔斯泰所写的《战争与和平》，它的地址是http://www.gutenberg.org/cache/epub/2600/pg2600.txt。

下面代码展示了如何下载（译者注：win上使用Git Bash）：

$ curl -sO http://www.gutenberg.org/cache/epub/2600/pg2600.txt
$ wc pg2600.txt
   65007  566320 3291648 pg2600.txt

接下来，我们统计love一词出现的次数，忽略大小写，如下所示（译者注：会有编码问题）：

$ time (grep -io love pg2600.txt | wc -l)
677
(grep -io love pg2600.txt) 0.11s user 0.00s system 98% cpu 0.116 total

现在使用Python的协程来做（grep.py）：

def coroutine(fn):
    def wrapper(*args, **kwargs):
        c = fn(*args, **kwargs)
        next(c)
        return c
    return wrapper


def cat(f, case_insensitive, child):
    if case_insensitive:
        line_processor = lambda l: l.lower()
    else:
        line_processor = lambda l: l

    for line in f:
        child.send(line_processor(line))


@coroutine
def grep(substring, case_insensitive, child):
    if case_insensitive:
        substring = substring.lower()
    while True:
        text = (yield)
        child.send(text.count(substring))


@coroutine
def count(substring):
    n = 0
    try:
        while True:
            n += (yield)
    except GeneratorExit:
        print(substring, n)


if __name__ == '__main__':
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument('-i', action='store_true',
                        dest='case_insensitive')
    parser.add_argument('pattern', type=str)
    parser.add_argument('infile', type=argparse.FileType('r'))

    args = parser.parse_args()

    cat(args.infile, args.case_insensitive,
        grep(args.pattern, args.case_insensitive,
             count(args.pattern)))

分析代码之前，我们先运行一下，和grep进行比较：

$ time python3.5 grep.py -i love pg2600.txt
love 677
python3.5 grep.py -i love pg2600.txt  0.09s user 0.01s system 97% cpu 0.097 total

可以看到，使用协程的纯Python版本与使用grep和wc命令的Unix相比，十分具有竞争力。当然，Unix的grep命令远比Python版本强大。不能简单宣称Python比C语言快！但是，Python的结果也是让人满意的。

来分析下代码。首先，再次执行coroutine的装饰器。之后，将总任务分解成三块：

逐行读取文件（通过cat函数）
统计每行中substring的出现次数（grep协程）
求和并打印数据（count协程）

在脚本文件的主体部分，我们解析命令行选项，将cat结果传给grep，将grep结果传给count，就像操作普通的Unix工具。

实现这个链条极其简单。我们将接收数据的协程当做参数（前面例子的child），传递给产生数据的函数或协程。然后，在数据源中，调用协程的send方法。

第一个函数cat，作为整个函数的数据源，它逐行读取文件，将每行发送给grep
（child.send(line)）。如果匹配是大小写不敏感的，不需要进行转换；如果大小写敏感，则都转化为小写。

grep命令是我们的第一个协程。这里，进入一个无限循环，持续获取数据（text = (yield)），统计substring在text中的出现次数，，将次数发送给写一个协程（即count）：child.send(text.count(substring)))。

count协程用总次数n，从grep获取数据，对总次数进行求和，n += (yield)。它捕获发送给各个协程关闭时的GeneratorExit异常（在我们的例子中，到达文件最后就会出现异常），以判断何时打印这个substring和n。

当把协程组织为更复杂的结构时，会更有趣。比如，我们可以统计多个单词出现的次数。

下面的代码展示了一种这样做的方法，通过一个额外的协程负责广播，将输入数据发送给任意数目的子协程（mgrep.py）：

def coroutine(fn):
    def wrapper(*args, **kwargs):
        c = fn(*args, **kwargs)
        next(c)
        return c
    return wrapper

def cat(f, case_insensitive, child):
    if case_insensitive:
        line_processor = lambda l: l.lower()
    else:
        line_processor = lambda l: l

    for line in f:
        child.send(line_processor(line))

@coroutine
def grep(substring, case_insensitive, child):
    if case_insensitive:
        substring = substring.lower()
    while True:
        text = (yield)
        child.send(text.count(substring))

@coroutine
def count(substring):
    n = 0
    try:
        while True:
            n += (yield)
    except GeneratorExit:
        print(substring, n)

@coroutine
def fanout(children):
    while True:
        data = (yield)
        for child in children:
            child.send(data)

if __name__ == '__main__':
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument('-i', action='store_true', dest='case_insensitive')
    parser.add_argument('patterns', type=str, nargs='+',)
    parser.add_argument('infile', type=argparse.FileType('r'))

    args = parser.parse_args()

    cat(args.infile, args.case_insensitive,
        fanout([grep(p, args.case_insensitive,
                     count(p)) for p in args.patterns]))

代码看上去和之前的例子差不多。让我们分析一下差别。我们定义了一个广播器：fanout。fanout()协程使用一列协程作为输入，自身位于一个无限循环中。当收到数据后（data = (yield)），便将数据分发给注册的协程（for child in children: child.send(data)）。

不用修改cat、grep、count的代码，我们就可以利用原有的代码来搜索任意个数的字符串了！

它的性能依旧很好，如下所示：

$ time python3.5 mgrep.py -i love hate hope pg2600.txt
hate 103
love 677
hope 158
python3.5 mgrep.py -i love hate hope pg2600.txt  0.16s user 0.01s system 98% cpu 0.166 total

总结

Python从1.5.2版本之后引入了asyncore和asynchat模块，开始支持异步编程。2.5版本引入了yield，可以向协程传递数据，简化了代码、加强了性能。Python 3.4 引入了一个新的库进行异步I/O，称作asyncio。

Python 3.5通过async def和await，引入了真正的协程类型。感兴趣的读者可以继续研究Python的新扩展。一句警告：异步编程是一个强大的工具，可以极大地提高I/O密集型代码的性能。但是异步编程也是存在问题的，而且还相当复杂。

任何异步代码都要精心选择非阻塞的库，以防使用阻塞代码。并且要运行一个协程规划期（因为OS不能像规划线程一样规划协程），包括写一个事件循环和其它事务。读异步代码会有一定困难，即使我们的最简单的例子也很难一眼看懂。所以，一定要小心！

《Python分布式计算》第2章异步编程（Distributed Computing with Python）

协程

一个异步实例

总结

热门文章

最新文章

相关课程

相关电子书

相关实验场景

推荐镜像

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

《Python分布式计算》第2章 异步编程 （Distributed Computing with Python）

协程

一个异步实例

总结

热门文章

最新文章

相关课程

相关电子书

相关实验场景

推荐镜像

《Python分布式计算》第2章异步编程（Distributed Computing with Python）