协程

阅读：273697661 分享到

通常在 Python 中我们进行并发编程一般都是使用多线程或者多进程来实现的，对于计算型任务由于 GIL 的存在我们通常使用多进程来实现，而对于 IO 型任务我们可以通过线程调度来让线程在执行 IO 任务时让出 GIL，从而实现表面上的并发。其实对于 IO 型任务我们还有一种选择就是协程，协程是运行在单线程当中的"并发"，协程相比多线程一大优势就是省去了多线程之间的切换开销，获得了更大的运行效率。

协程，又称微线程，纤程，英文名 Coroutine。协程的作用是在执行函数 A 时可以随时中断去执行函数 B，然后中断函数 B 继续执行函数 A（可以自由切换）。但这一过程并不是函数调用，这一整个过程看似像多线程，然而协程只有一个线程执行。

协程可以处理 IO 密集型程序的效率问题，但是处理 CPU（计算）密集型不是它的长处，如要充分发挥 CPU 利用率可以结合多进程+协程。

asyncio + yield from

asyncio 是 Python3.4 版本引入的标准库，直接内置了对异步 IO 的支持。asyncio 的异步操作，需要在 coroutine 中通过 yield from 完成。看如下代码：

import asyncio

@asyncio.coroutine
def test(i):
    print('test1', i)
    yield from asyncio.sleep(1)
    print('test2', i)


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    tasks = [test(i) for i in range(3)]
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()

@asyncio.coroutine 把一个 generator 标记为 coroutine 类型，然后就把这个 coroutine 扔到 EventLoop 中执行。test() 会首先打印出test1，然后 yield from 语法可以让我们方便地调用另一个 generator。由于 asyncio.sleep() 也是一个 coroutine，所以线程不会等待 asyncio.sleep()，而是直接中断并执行下一个消息循环。当 asyncio.sleep() 返回时，线程就可以从 yield from 拿到返回值（此处是 None），然后接着执行下一行语句。把 asyncio.sleep(1) 看成是一个耗时 1 秒的 IO 操作，在此期间主线程并未等待，而是去执行 EventLoop 中其他可以执行的 coroutine 了，因此可以实现并发执行。

asyncio + async/await

为了简化并更好地标识异步 IO，从 Python3.5 开始引入了新的语法 async 和 await，可以让 coroutine 的代码更简洁易读。请注意，async 和 await 是 coroutine 的新语法，使用新语法只需要把 @asyncio.coroutine 替换为 async 和把 yield from 替换为 await。

import asyncio

async def test(i):
    print('test1', i)
    await asyncio.sleep(1)
    print('test2', i)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    tasks = [test(i) for i in range(3)]
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()

运行结果与之前一致。与前面的代码相比，这里只是把 yield from 换成了await，@asyncio.coroutine 换成了 async，其余不变。

gevent

gevent 是一个基于 greenlet 实现的网络库，通过 greenlet 实现协程。基本思想是一个 greenlet 就认为是一个协程，当一个 greenlet 遇到 IO 操作的时候，比如访问网络，就会自动切换到其他的 greenlet，等到 IO 操作完成，再在适当的时候切换回来继续执行。由于 IO 操作非常耗时，经常使程序处于等待状态，有了 gevent 为我们自动切换协程，就保证总有 greenlet 在运行，而不是等待 IO 操作。

gevent 是第三方模块，所有我们首先安装上该模块，安装命令：pip install gevent 。

import gevent

def test(n):
    for i in range(n):
        print(gevent.getcurrent(), i)

if __name__ == '__main__':
    g1 = gevent.spawn(test, 3)
    g2 = gevent.spawn(test, 3)
    g3 = gevent.spawn(test, 3)

    g1.join()
    g2.join()
    g3.join()

运行结果：

<Greenlet at 0x10ff1e0: test(3)> 0
<Greenlet at 0x10ff1e0: test(3)> 1
<Greenlet at 0x10ff1e0: test(3)> 2
<Greenlet at 0x10ffae0: test(3)> 0
<Greenlet at 0x10ffae0: test(3)> 1
<Greenlet at 0x10ffae0: test(3)> 2
<Greenlet at 0x10ffb70: test(3)> 0
<Greenlet at 0x10ffb70: test(3)> 1
<Greenlet at 0x10ffb70: test(3)> 2

可以看到 3 个 greenlet 是依次运行而不是交替运行。要让 greenlet交替运行，可以通过 gevent.sleep() 交出控制权：

import gevent

def test(n):
    for i in range(n):
        print(gevent.getcurrent(), i)
        gevent.sleep(1)

if __name__ == '__main__':
    g1 = gevent.spawn(test, 3)
    g2 = gevent.spawn(test, 3)
    g3 = gevent.spawn(test, 3)

    g1.join()
    g2.join()
    g3.join()

运行结果：

<Greenlet at 0xec6150: test(3)> 0
<Greenlet at 0xec6a50: test(3)> 0
<Greenlet at 0xec6ae0: test(3)> 0
<Greenlet at 0xec6150: test(3)> 1
<Greenlet at 0xec6a50: test(3)> 1
<Greenlet at 0xec6ae0: test(3)> 1
<Greenlet at 0xec6150: test(3)> 2
<Greenlet at 0xec6a50: test(3)> 2
<Greenlet at 0xec6ae0: test(3)> 2

当然在实际的代码里，我们不会用 gevent.sleep() 去切换协程，而是在执行到 IO 操作时 gevent 会自动完成，所以 gevent 需要将 Python 自带的一些标准库的运行方式由阻塞式调用变为非阻塞式运行。这一过程在启动时通过 monkey patch 完成：

from urllib import request
import gevent
from gevent import monkey
monkey.patch_all()

def test(url):
    print('Get: %s' % url)
    response = request.urlopen(url)
    content = response.read().decode('utf8')
    print('%d bytes received from %s.' % (len(content), url))

if __name__ == '__main__':
    gevent.joinall([
        gevent.spawn(test, 'http://httpbin.org/ip'),
        gevent.spawn(test, 'http://httpbin.org/uuid'),
        gevent.spawn(test, 'http://httpbin.org/user-agent')
    ])

运行结果：

Get: https://www.birdpython.com/
Get: https://www.birdpython.com/columns/1/
Get: https://www.birdpython.com/video/
55450 bytes received from https://www.birdpython.com/.
35775 bytes received from https://www.birdpython.com/video/.
55682 bytes received from https://www.birdpython.com/columns/1/.

从结果看，3 个网络操作是并发执行的，而且结束顺序不同，但只有一个线程。

思考

首先，我们知道在多线程中，CPU 可以对每个线程分配时间片，CPU 本身做线程之间的调度（切换），我们每个线程中的阻塞行为不会影响到其它线程继续执行代码。

目前为止，喜欢思考的同学可能会想到一个问题：对于单线程中的阻塞行为，协程是如何实现调度的？也就是说，协程既然是单线程的，他是如何在阻塞之间切换的呢？

其实，所有的协程库，比如我们刚刚学习的 asyncio 和 gevent 等等，他们只是把阻塞行为的类库修改成了非阻塞行为的类库而已。比如，我们以阻塞套接字为例：当我们用 ss = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 定义一个阻塞套接字，当用了协程库 gevent，你只需要加上 from gevent import monkey 和 monkey.patch_all() 两行代码，协程库就相当于给我们的套接字加上一行代码 setblocking(False)，该代码就会让套接字变成非阻塞的套接字了，这样一来，当程序执行原来的阻塞函数，比如 accept，recv 等等，就会马上挂起，当有其它任务时，就会去执行其它任务了，这样就实现了单线程并发。

比如，python 的 sleep 函数也是阻塞的，如果使用协程库 gevent，并且执行代码 from gevent import monkey 和 monkey.patch_all() 后，阻塞的 sleep 就变成了非阻塞的函数，当程序执行到 sleep 代码时，程序就不会阻塞，gevent 就会挂起，去调度其它任务。

注意，协程库只能动态的修改 Python 自带的一些标准库，可能有同学会问，为什么协程库 gevent 能把 requests 这种第三方库也修改成非阻塞的呢？那是因为 requests 库本身用的就是基于 Pyhton 自带的标准库 socket 库啊！

但是，协程库并不是把所有的阻塞类库的行为变成非阻塞的行为，比如对 python 的阻塞函数 input 使用协程库，就不会起作用。

import gevent
from gevent import monkey
monkey.patch_all()

def test(name):
   print(name)
   input("我阻塞，你协程库能奈我何...")

if __name__ == '__main__':
    gevent.joinall([
        gevent.spawn(test, "任务1"),
        gevent.spawn(test, "任务2"),
        gevent.spawn(test, "任务3")
    ])

有些同学可能会想，为什么协程库没有对 input 做非阻塞化呢，那是因为 input 本身就没有非阻塞的 API 的啊！协程库自己也没办法啊！

同样，对于 CPU（计算）密集型导致的阻塞行为，协程也是无能为力，如下代码。

import gevent
from gevent import monkey
monkey.patch_all()

def test(name):
   print(name)
   print("我下面做计算导致的阻塞，你协程库也没啥卵用啊。。。")
   while True:  # CPU（计算）密集型
       1 + 1

if __name__ == '__main__':
    gevent.joinall([
        gevent.spawn(test, "任务1"),
        gevent.spawn(test, "任务2"),
        gevent.spawn(test, "任务3")
    ])

至此 Python 中的协程就介绍完毕了，在实际项目中可以使用协程异步的读写网络、读写文件、渲染界面等，而在等待协程完成的同时，CPU 还可以进行其他的计算，协程的作用正在于此。那么协程和多线程的差异在哪里呢？多线程的切换需要靠操作系统来完成，当线程越来越多时切换的成本会很高，而协程是在一个线程内切换的，切换过程由我们自己控制，因此开销小很多，这就是协程和多线程的根本差异。