协程和并发课程【部分一~二】

之前偶然在博文 Yield 和 Coroutine 发现了作者推荐的课程 A Curious Course on Coroutines and Concurrency，才发现原来自己对 yield 的强大实在理解的太少。这个课程的作者是 David Beazley。他是 Python 拥趸（Pythonista）； 《Python Essential Reference》 作者；Swig、PLC 等包作者；全职 Python 培训师。大牛！其课程写的实在太好了，赞赞赞！作者还提供了该课程详细的 PPT 版本（198 页，全英文…）。课程大纲如下图：

可以看到课程是越来越难的，特别是最后一个，对操作系统还得有一定的了解。包括代码和 PPT 的打包链接在课程主页上有，但为了保险起见，我额外将其备份至百度网盘和腾讯微云。

本文将主要对该 PPT 课程重新整理，会很长，所以会分成两篇文章。

对于 yield 和协程基础，可参考前两篇博文：

课程预备

在课程中，作者多次以对日志的处理作为例子，所以这里我们先看一下如何生成日志吧。很简单，只要执行 logsim.py 文件就行了，如下图：

第一部分：生成器和协程介绍

生成器

对于生成器，我们在博文祥析 yield 关键字和生成器里有详细介绍。简要来说，生成器就是调用了 yield 关键字的函数，本质上是一个迭代器，用于生成序列。

倒计时

倒计时的程序还是比较简单的：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# countdown.py
#
# A simple generator function


def countdown(n):
    print "Counting down from", n
    while n > 0:
        yield n
        n -= 1
    print "Done counting down"


# Example use
if __name__ == '__main__':
    for i in countdown(10):
        print i

tail -f

作者用生成器实现了 Unix/Linux 中的 Python 版 tail -f：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# follow.py
#
# A generator that follows a log file like Unix 'tail -f'.
#
# Note: To see this example work, you need to apply to 
# an active server log file.  Run the program "logsim.py"
# in the background to simulate such a file.  This program
# will write entries to a file "access-log".

import time


def follow(thefile):
    thefile.seek(0, 2)  # Go to the end of the file
    while True:
        line = thefile.readline()
        if not line:
            time.sleep(0.1)  # Sleep briefly
            continue
        yield line

# Example use
if __name__ == '__main__':
    logfile = open("access-log")
    for line in follow(logfile):
        print line,

借助 Tmux，可以看到如下效果：

流水线（Pipeline）

生成器另一个重要的功能就是实现流水线了，如下图：

接下来，我们可以仿照 Unix/Linux 上管道思想，对 tail -f 的数据进行过滤（grep）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# pipeline.py
#
# An example of setting up a processing pipeline with generators


def grep(pattern, lines):
    for line in lines:
        if pattern in line:
            yield line


if __name__ == '__main__':
    from follow import follow

    # Set up a processing pipe : tail -f | grep python
    logfile = open("access-log")
    loglines = follow(logfile)
    pylines = grep("python", loglines)

    # Pull results out of the processing pipeline
    for line in pylines:
        print line,

效果图如下：

协程

在之前博文 Python 协程我们对协程有一个详细探讨。协程本质上是执行权的转移，从而实现协同工作（cooperative routine），它的含义不单单包括了生成器，还涵盖了协同工作；而这些生成器是没有的，它只是一个独立的函数。本课程作者就说的表层但容易理解一点，也即，生成器用于生成数据，而协程则倾向于消费数据。

另外一点，作者提到的非常重要的是，协程跟迭代没有关系（Coroutines are not related to iteration）。协程更像是一个流水线，一级一级传递执行权。

grep

作者提供了一个很简单的协程，模仿 Unix/Linux 的 grep 函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# grep.py
#
# A very simple coroutine


def grep(pattern):
    print "Looking for %s" % pattern
    while True:
        line = (yield)
        if pattern in line:
            print line,


# Example use
if __name__ == '__main__':
    g = grep("python")
    g.next()
    g.send("Yeah, but no, but yeah, but no")
    g.send("A series of tubes")
    g.send("python generators rock!")

注：这里的 grep 函数因为执行权转移是通过 yield 表达式和 send() 方法做到的，所以它是一个协程，而不是一个生成器。而且它是对发送过来的数据进行了消费（处理）。

协程装饰器

我们知道，在给协程发送数据之前，要先“启动”它，以便于该协程执行到第一个 yield 表达式处。启动的方式是调用 next() 或者 send(None)。详细可参考上一博文 Python 协程。

为此，作者干脆就提供了一个协程装饰器，方便启动协程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
# coroutine.py
#
# A decorator function that takes care of starting a coroutine
# automatically on call.


def coroutine(func):
    def start(*args, **kwargs):
        cr = func(*args, **kwargs)
        cr.next()
        return cr

    return start


# Example use
if __name__ == '__main__':
    @coroutine
    def grep(pattern):
        print "Looking for %s" % pattern
        while True:
            line = (yield)
            if pattern in line:
                print line,


    g = grep("python")
    # Notice how you don't need a next() call here
    g.send("Yeah, but no, but yeah, but no")
    g.send("A series of tubes")
    g.send("python generators rock!")

后边例子都会用到该装饰器。

协程关闭及异常处理

关于协程关闭即异常处理，可以参考 PEP 342 – Coroutines via Enhanced Generators。

协程关闭函数是 close()，样例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# grepclose.py
#
# A coroutine that catches the close() operation

from coroutine import coroutine


@coroutine
def grep(pattern):
    print "Looking for %s" % pattern
    try:
        while True:
            line = (yield)
            if pattern in line:
                print line,
    except GeneratorExit:
        print "Going away. Goodbye"


# Example use
if __name__ == '__main__':
    g = grep("python")
    g.send("Yeah, but no, but yeah, but no\n")
    g.send("A series of tubes\n")
    g.send("python generators rock!\n")
    g.close()

古怪的例子

关于该例子详细解析请参考之前博文 Python 协程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# bogus.py
#
# Bogus example of a generator that produces and receives values


def countdown(n):
    print "Counting down from", n
    while n >= 0:
        newvalue = (yield n)
        # If a new value got sent in, reset n with it
        if newvalue is not None:
            n = newvalue
        else:
            n -= 1


# The holy grail countdown
c = countdown(5)
for x in c:
    print x,
    if x == 5:
        c.send(3)

第二部分：协程、流水线和数据流

流水线

流水线需要 3 个部分：

  1. 源头（soure）：数据源，生产者，驱动整个流水线

    代码样例（源头不是一个协程）：

  2. 管道（pipe）：协程用于建立数据处理流水线

  3. 终点（sink）：收集数据并处理，相当于最终消费者

    代码样例：

其中，源头不是协程，管道和终点都是协程。

tail -f

我们先来看一个只有源头和终点的例子（注意源头不是协程）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
# cofollow.py
#
# A simple example showing how to hook up a pipeline with
# coroutines.   To run this, you will need a log file.
# Run the program logsim.py in the background to get a data
# source.

from coroutine import coroutine

# A data source.  This is not a coroutine, but it sends
# data into one (target)

import time


def follow(thefile, target):
    thefile.seek(0, 2)  # Go to the end of the file
    while True:
        line = thefile.readline()
        if not line:
            time.sleep(0.1)  # Sleep briefly
            continue
        target.send(line)


# A sink.  A coroutine that receives data

@coroutine
def printer():
    while True:
        line = (yield)
        print line,


# Example use
if __name__ == '__main__':
    f = open("access-log")
    follow(f, printer())

过滤器

为上个例子添加一个协程过滤器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
# copipe.py
#
# A simple example showing how to hook up a pipeline with
# coroutines.   To run this, you will need a log file.
# Run the program logsim.py in the background to get a data
# source.

from coroutine import coroutine

# A data source.  This is not a coroutine, but it sends
# data into one (target)

import time


def follow(thefile, target):
    thefile.seek(0, 2)  # Go to the end of the file
    while True:
        line = thefile.readline()
        if not line:
            time.sleep(0.1)  # Sleep briefly
            continue
        target.send(line)


# A filter.

@coroutine
def grep(pattern, target):
    while True:
        line = (yield)  # Receive a line
        if pattern in line:
            target.send(line)  # Send to next stage


# A sink.  A coroutine that receives data

@coroutine
def printer():
    while True:
        line = (yield)
        print line,


# Example use
if __name__ == '__main__':
    f = open("access-log")
    follow(f,
           grep('python',
                printer()))

生成器和协程流水线的区别

生成器和协程流水线的示意图如下：

它们之间关键的区别在于：生成器通过迭代从管道中拉取（pull）数据；协程用 send() 将数据推送（push）至管道。一个主动，一个被动。

广播

利用协程，我们可以将数据发送到多个地方，如下图所示：

一个广播的例子如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
# cobroadcast.py
#
# An example of broadcasting a data stream onto multiple coroutine targets.

from coroutine import coroutine

# A data source.  This is not a coroutine, but it sends
# data into one (target)

import time


def follow(thefile, target):
    thefile.seek(0, 2)  # Go to the end of the file
    while True:
        line = thefile.readline()
        if not line:
            time.sleep(0.1)  # Sleep briefly
            continue
        target.send(line)


# A filter.
@coroutine
def grep(pattern, target):
    while True:
        line = (yield)  # Receive a line
        if pattern in line:
            target.send(line)  # Send to next stage


# A sink.  A coroutine that receives data
@coroutine
def printer():
    while True:
        line = (yield)
        print line,


# Broadcast a stream onto multiple targets
@coroutine
def broadcast(targets):
    while True:
        item = (yield)
        for target in targets:
            target.send(item)


# Example use
if __name__ == '__main__':
    f = open("access-log")
    follow(f,
           broadcast([grep('python', printer()),
                      grep('ply', printer()),
                      grep('swig', printer())])
           )

可以修改该例子，用同一个 printer 来打印所有广播结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# cobroadcast2.py
#
# An example of broadcasting a data stream onto multiple coroutine targets.
# This example shows "fan-in"---a situation where multiple coroutines
# send to the same target.

......
# Example use
if __name__ == '__main__':
    f = open("access-log")
    p = printer()
    follow(f,
           broadcast([grep('python', p),
                      grep('ply', p),
                      grep('swig', p)])
           )

协程与对象

协程跟面向对象设计（OO Design）版本的处理方式有些类似：

1
2
3
4
5
6
7
8
class GrepHandler(object):
    def __init__(self, pattern, target):
        self.pattern = pattern
        self.target = target

    def send(self, line):
        if self.pattern in line:
            self.target.send(line)

相对应的协程版本为：

1
2
3
4
5
6
@coroutine
def grep(pattern, target):
    while True:
        line = (yield)
        if pattern in line:
            target.send(line)

相较之下，协程定义更为简洁（背后的工作就不一定简单了），只是一个函数，而面向对象版本的就比较复杂，需要类定义、方法定义，可能还需要基类和库。

另外，作者还对这两者的处理速度进行了测试：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
# benchmark.py
#
# A micro benchmark comparing the performance of sending messages into
# a coroutine vs. sending messages into an object


# An object
class GrepHandler(object):
    def __init__(self, pattern, target):
        self.pattern = pattern
        self.target = target

    def send(self, line):
        if self.pattern in line:
            self.target.send(line)


# A coroutine
from coroutine import coroutine


@coroutine
def grep(pattern, target):
    while True:
        line = (yield)
        if pattern in line:
            target.send(line)


# A null-sink to send data
@coroutine
def null():
    while True:
        item = (yield)


# A benchmark
line = ''
for _ in range(1000000):
    line += 'python is nice\n'
p1 = grep('python', null())  # Coroutine
p2 = GrepHandler('python', null())  # Object

from timeit import timeit

print "coroutine:", timeit("p1.send(line)",
                           "from __main__ import line, p1")

print "object:", timeit("p2.send(line)",
                        "from __main__ import line, p2")

作者的测试结果为：

注：当数据量较小时，协程不一定比类的快。

作者对结果进行了分析，认为主要问题在于对于类实例而言，查找某个变量会消耗时间：

课程的其他部分见后续博文。

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

谢宏峰，毕业于暨南大学（2009 - 2013）、中国科学院（2013 - 2016），现就职于深信服科技，从事 OpenStack 开发。