Python 协程

在学习 Python 的协程之前,先要了解 Python 的 List Comprehensions 和 Generators。

List Comprehensions

List Comprehensions 一般被称为列表生成式,使用它可以用来创建 list 集合。

生成简单的 list,比如 [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],可以使用 list(range(1, 11)) 来实现。对于复杂的,比如 [1*1, 2*2, 3*3, ..., 10*10],我们就可以通过列表生成式来生成,写成 [x * x for x in range(1, 11)]

在列表生成式中还可以加入条件判断,满足条件的才会生成,比如:[x * x for x in range(1, 11) if x % 2 == 0] 会仅生成偶数的平方 list 集合。

由于 for 循环其实可以同时使用两个甚至多个变量,比如 dict 的 items() 方法可以同时迭代 key 和 value,所以列表生成式也可以使用两个变量来生成 list,比如:[k + "=" + v for k, v in {'x': 'X', 'y': 'Y'}.items()]

Generators

使用列表生成式可以很方便地创建一个列表,但是这个列表的容量肯定是受限的,不能无限大,并且如果只使用列表前面的一部分数据,那么分配的大部分内存空间就被浪费掉了。Python 的生成器采用了另外一种思路,即并不预先创建出所有的数据,而是只将创建元素的逻辑提前编写好,在需要使用元素时,通过该逻辑计算出元素的值。

在 Python 中,创建 Generators 主要有两种方式。

第一种方式很简单,将列表生成式的方括号改为圆括号即可。

1
2
3
4
5
6
7
8
9
# 列表生成式
l = [x * x for x in range(1, 11)]
# 生成器
g = (x * x for x in range(1, 11))
# 获取生成器的下一个值
n = next(g)
# 迭代
for n in g:
print(n)

当推算的算法比较复杂时,可以用第二种方式,也就是函数的方式去实现。

比如斐波那契数列,第一个数和第二个数都是 1,除了它俩,其余任意一个数都可由前两个数相加得到。

1
2
3
4
5
6
def fib(num):
i, a, b = 0, 0, 1
while i < num:
print(b, end=' ')
a, b = b, a + b
i = i + 1

这种由前面的元素推算后续元素的逻辑非常类似生成器,此时只要将函数中的打印值改为 yield b 即可将函数变成生成器。

1
2
3
4
5
6
7
8
9
10
11
12
13
import types

def fib(num):
i, a, b = 0, 0, 1
while i < num:
yield b
a, b = b, a + b
i = i + 1

# <class 'generator'>
print(type(fib(11)))
# True
print(isinstance(fib(11), types.GeneratorType))

在函数中,代码是顺序执行的,遇到 return 语句或者执行完毕后返回;而生成器在每次调用 next() 时执行,遇到 yield 语句返回,下次执行时会从上次返回的 yield 语句处继续执行。

什么是协程

协程也叫做微线程,纤程,英文名为 Coroutine,可以看做是用户态的轻量级线程。

在说协程之前,我们先简单回忆一下进程和线程。进程可以简单理解为程序的启动实例,比如运行一个游戏,打开一个软件,就是开启了一个进程。而线程通常从属于进程,是程序的实际执行者。对于操作系统来说,线程是最小的执行单元,进程是最小的资源管理单元。

协程在外观上看就是一段函数,但是在执行过程中,函数内部是可以中断的,然后转而去执行别的函数(并不是函数调用),并在适当的时候再返回来接着执行。从现象上看,有点类似 CPU 的中断,比如:

1
2
3
4
5
6
7
8
9
10
11
12
def printNum():
print(1)
print(2)
print(3)

def printWord():
print('a')
print('b')
print('c')

printNum()
printWord()

假如上述代码通过协程去执行,那么 printNum() 在执行过程中随时可以中断去执行 printWord(),结果可能为:

1
2
3
4
5
6
1
2
a
b
3
c

这个结果看起来像是使用了多线程,然而实际上协程使用的是单个线程。

协程与进程和线程的区别

协程与进程和线程的根本区别是,进程和线程是操作系统级别的,而协程是编译器级别的。虽然进程和线程在语言层次都有体现,但本质上都是操作系统提供的 API,程序的执行依赖操作系统的调度算法,每次暂停的位置是不确定的,也就意味着重新开始的位置不可预知;而协程是编译器的魔术,通常插入相关的代码使得代码段能够实现分段式的执行,每次执行遇到 yield 返回,重新开始时会从上一次的 yield 处继续向下执行。

协程的优势

  • 不存在线程切换开销
    因为协程是函数切换而不是线程切换,切换的位置可以由程序自身控制,因此也就没有线程切换的开销。和多线程比,线程数量越多,协程的性能优势就越明显。

  • 不存在线程安全问题
    由于不是多线程,也就不需要多线程的锁机制,因为只有一个线程,也就不存在同时写变量冲突,在协程中控制共享资源不加锁,只需要判断状态就好了,所以执行效率比多线程高很多。

协程的劣势

  • 无法发挥多核性能
    由于协程使用的是单个线程,所以它无法充分发挥 CPU 的多核性能。当然我们可以通过使用多进程 + 协程的方式来充分利用多核,我们日常编写的大部分代码都没有必要这么做,除非是在 CPU 密集型应用中。

  • 协程阻塞时会阻塞整个程序
    协程在遇到阻塞操作,如 IO 操作时会阻塞掉整个程序,因此我们使用协程时一般会配合 Event Loop 进行异步操作。

Python 的实现

在 Python 2.x 中,协程主要通过生成器配合 yield 和 send 来实现,或者使用第三方库,比如 greenlet,甚至是 gevent

在 Python 3.3 中,官方加入了 yield from 语法,在 3.4 版本中还提供了支持异步 IO 的标准库 asyncio,在 3.5 版本中,加入了基于 asyncio 的新语法 async/await。因此在 Python 3.x 中实现协程可以有多种组合方式,比如:

  • 在 Python 3.4 版本中,可以通过 asyncio + yield from 的方式实现
  • 在 Python 3.5 版本中,可以通过 async/await 实现
  • 使用第三方库 gevent

协程在运行的过程中共有四种状态,GEN_CREATE(等待执行)、GEN_RUNNING(正在执行)、GEN_SUSPENDED(中断执行) 和 GEN_CLOSED(执行结束)。这个状态可以通过 getgeneratorstate() 方法获取,在调用生成器生成对象后,此时的协程处于 GEN_CREATE 状态,即等待开始的状态。接下来需要预激协程,此时可以通过 next() 或者 send(None) 预激。预激之后的协程开始执行,当运行到 yield 处会中断执行,此时的状态就是 GEN_SUSPENDED

基于生成器实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
#!/usr/bin/env python3
# -*- coding: utf-8 -*-


def simple_coroutine():
print('coroutine started')
x = yield
print('coroutine received: ', x)


if __name__ == '__main__':
from inspect import getgeneratorstate

m = simple_coroutine()
# 获取协程的状态
print(getgeneratorstate(m))
# 使用 next 方法预激
print(next(m))
# 发送数据
m.send(12)

在使用 Generators 实现协程时,yield 通常出现在表达式的右边,比如 x = yield。yield 可以产出值,也可以不产出值。如果 yield 右边没有表达式,那么值为 None,比如 x = yield 产出的值就是 None,而 x = yield b 产出的值为 b 的值。同时在调用方可以通过 send() 方法发送数据给协程,发送的数据会成为 yield 表达式的值。

终止协程和异常处理

在 Python 2.5 之后,生成器对象添加了 throw 和 close 方法,throw 方法会让生成器在暂停的 yield 表达式处抛出指定的异常,如果生成器处理了该异常,那么代码会继续执行到下一个 yield 表达式。如果没有处理该异常,或者抛出了另一个异常,那么异常会向上传递到调用方。

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
#!/usr/bin/env python3
# -*- coding: utf-8 -*-


class TestException(Exception):
pass


def demo_test_throw(b):
print('coroutine started')
while True:
try:
x = yield b
except TestException:
print('TestException handled')
else:
print('coroutine received: ', x)
print('测试 throw 后代码的执行情况')


if __name__ == '__main__':
m = demo_test_throw(1)
print(next(m))
# 抛出自定义异常
m.throw(TestException)
print(next(m))
"""
coroutine started
1
TestException handled
测试 throw 后代码的执行情况
coroutine received: None
测试 throw 后代码的执行情况
1
"""

在生成器因为没有后续语句而退出时,如果没有 yield 值产生,则会抛出一个 StopIteration 异常。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
#!/usr/bin/env python3
# -*- coding: utf-8 -*-


class TestException(Exception):
pass


def demo_test_throw(b):
print('coroutine started')
try:
x = yield b
except TestException:
print('TestException handled')
else:
print('coroutine received: ', x)


if __name__ == '__main__':
m = demo_test_throw(1)
print(next(m))
# 抛出自定义异常
m.throw(TestException)
# 最终会抛出一个 StopIteration 异常

close 方法会让生成器在暂停的 yield 表达式处抛出 GeneratorExit 异常,如果生成器没有处理这个异常,或者抛出了 StopIteration 异常,则调用方不会报错。如果抛出了其他异常,则会向上传递给调用方。在抛出 GeneratorExit 异常后,生成器一定不能产生值,否则会抛出 RuntimeError 异常。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
#!/usr/bin/env python3
# -*- coding: utf-8 -*-


def demo_test_close(b):
print('coroutine started')
try:
x = yield b
except GeneratorExit:
print('coroutine closed')
# 这里加入 yield 表达式会抛出 RuntimeError
else:
print('coroutine received: ', x)


if __name__ == '__main__':
m = demo_test_close(1)
print(next(m))
m.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
28
29
30
31
32
33
34
35
36
37
38
39
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from functools import wraps


def coroutine(fn):
"""
装饰器,预激 fn
:param fn:
:return:
"""
# 把装饰器生成器函数替换成这里的 primer
# 函数调用 primer 函数时,返回预激后的生成器
@wraps(fn)
def primer(*args, **kwargs):
# 调用被装饰的函数,获取生成器对象
generator = fn(*args, **kwargs)
# 预激生成器
next(generator)
# 返回生成器
return generator
return primer


@coroutine
def simple_coroutine():
print('coroutine started')
x = yield
print('coroutine received: ', x)


if __name__ == '__main__':
from inspect import getgeneratorstate

m = simple_coroutine()
# 获取协程的状态
print(getgeneratorstate(m))
# 发送数据
m.send(12)

返回值

在 Python 3 中,生成器可以返回值,在协程结束后(通过 return 正常返回,而不是主动停止),调用方可以通过捕获 StopIteration 异常来获取返回值。

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
#!/usr/bin/env python3
# -*- coding: utf-8 -*-


def demo_test_return():
total = 0
while True:
tmp = yield
if tmp is None:
break
total += tmp
return total


if __name__ == '__main__':
result = 0
m = demo_test_return()
next(m)
m.send(10)
try:
m.send(None)
except StopIteration as e:
result = e.value
finally:
print(result)

yield from

yield from 是从 Python 3.3 开始添加的语法,在其他语言中,类似的结构使用 await 关键字。在 yield from 的后面需要添加的是可迭代对象,它可以是普通的可迭代对象,也可以是迭代器,甚至是生成器。

我们可以通过一个例子简单认识 yield 和 yield from 的区别。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
#!/usr/bin/env python3
# -*- coding: utf-8 -*-


def demo_test_yield_from(*args):
for item in args:
for i in item:
yield i


if __name__ == '__main__':
s = 'AB'
li = [1, 2, 3]
d = {"name": "saber", "age": 22}
g = (i for i in range(1, 5))

generator = demo_test_yield_from(s, li, d, g)
print(list(generator))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
#!/usr/bin/env python3
# -*- coding: utf-8 -*-


def demo_test_yield_from(*args):
for item in args:
yield from item


if __name__ == '__main__':
s = 'AB'
li = [1, 2, 3]
d = {"name": "saber", "age": 22}
g = (i for i in range(1, 5))

generator = demo_test_yield_from(s, li, d, g)
print(list(generator))

更复杂的,我们可以在 yield from 后面加入子生成器来实现生成器的嵌套。当然使用 yield 也可以完成生成器的嵌套,但是使用 yield from 可以避免自己处理很多意料不到的异常,从而专注于业务实现。

在生成器的嵌套中,有几个重要的概念。比如,委派生成器指的是包含 yield from 表达式的生成器函数,子生成器指的是 yield from 后边添加的生成器函数,而调用方指的是调用委派生成器的客户端。

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
#!/usr/bin/env python3
# -*- coding: utf-8 -*-


# 子生成器
def calculate_sum():
total = 0
while True:
tmp = yield
if tmp is None:
break
total += tmp
return total


# 委派生成器
def proxy_generator(results):
while True:
results['total'] = yield from calculate_sum()


# 调用方
def main():
results = {}
cal_sum = proxy_generator(results)
# 预激
next(cal_sum)
cal_sum.send(10)
cal_sum.send(22)
cal_sum.send(14)
cal_sum.send(None)
print(results['total'])


if __name__ == '__main__':
main()

委派生成器的作用是在调用方和子生成器之间建立一个双向通道。调用方可以通过 send() 方法直接发送消息给子生成器,而子生成器产生的 yield 的值也可以直接返回给调用方。

yield from 的具体语义很难理解,我们为什么还要使用它呢?比如我们需要返回值,完全可以麻烦点通过异常的方式获取。其实 yield from 在背后为我们做了很多事,具体做了什么可以参考下面官方提供的说明。比如 RESULT = yield from EXPR 语句,解释器在背后做了哪些工作呢?

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
_i = iter(EXPR)
try:
_y = next(_i)
except StopIteration as _e:
_r = _e.value
else:
while 1:
try:
_s = yield _y
except GeneratorExit as _e:
try:
_m = _i.close
except AttributeError:
pass
else:
_m()
raise _e
except BaseException as _e:
_x = sys.exc_info()
try:
_m = _i.throw
except AttributeError:
raise _e
else:
try:
_y = _m(*_x)
except StopIteration as _e:
_r = _e.value
break
else:
try:
if _s is None:
_y = next(_i)
else:
_y = _i.send(_s)
except StopIteration as _e:
_r = _e.value
break
RESULT = _r

greenlet

使用 yield 可以实现协程,但是实现的过程不易于理解,我们可以使用第三方库 greenlet 来实现相同的效果。

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
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from greenlet import greenlet
import time


def test_demo1():
while True:
print('inner demo 01')
time.sleep(1)
g2.switch()


def test_demo2():
while True:
print('inner demo 02')
time.sleep(1)
g1.switch()


if __name__ == '__main__':
# 创建协程
g1 = greenlet(test_demo1)
g2 = greenlet(test_demo2)

# 手动切换至 g1,先启动 g1
g1.switch()

asyncio

asyncio 是 Python 3.4 版本引入的标准库,它支持异步 IO 操作。asyncio 的编程模型就是一个消息循环,类似 JavaScript 中的 Event Loop,我们将需要执行的协程放入 EventLoop 中就可以实现异步操作。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import asyncio
import threading


@asyncio.coroutine
def hello_world():
print("Hello World! (%s)" % threading.current_thread())
# 模拟耗时操作
r = yield from asyncio.sleep(1)
print("Hello Python! (%s)" % threading.current_thread())


if __name__ == '__main__':
# 获取 EventLoop
loop = asyncio.get_event_loop()
# 执行协程
loop.run_until_complete(asyncio.wait([hello_world(), hello_world()]))
loop.close()

@asyncio.coroutine把一个 generator 标记为 coroutine 类型,然后放到 EventLoop 中去执行。由于 asyncio.sleep() 也是一个协程,所以我们可以使用 yield from 接收,但是主线程并不会一直等待它执行完毕,而是直接中断并去执行下一个消息循环。当 asyncio.sleep() 返回时,主线程就可以从 yield from 处拿到返回值(此处是 None),然后继续向下执行。

async/await

为了更好的标识异步 IO,从 Python 3.5 开始引入了新的语法 asyncawait,使用该语法只需要将 @asyncio.coroutine 替换为 async,再把 yield from 替换为 await 即可。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import asyncio
import threading


async def hello_world():
print("Hello World! (%s)" % threading.current_thread())
# 模拟耗时操作
r = await asyncio.sleep(1)
print("Hello Python! (%s)" % threading.current_thread())


if __name__ == '__main__':
# 获取 EventLoop
loop = asyncio.get_event_loop()
# 执行协程
loop.run_until_complete(asyncio.wait([hello_world(), hello_world()]))
loop.close()

参考

在 Unity 中 StartCoroutine/yield return 这个模式到底是怎么应用的?其中的原理是什么?

Python 教程

PEP 380 – Syntax for Delegating to a Subgenerator

《流畅的 Python》