Python多任务之协程

2019 年 10 月 15 日
筆記

前言

协程的核心点在于协程的使用，即只需要了解怎么使用协程即可；但如果你想了解协程是怎么实现的，就需要了解依次了解可迭代，迭代器，生成器了；

如果你只想看协程的使用，那么只需要看第一部分内容就行了；如果如果想理解协程，可以按照顺序依次阅读本博文，或者按照迭代器-生成器-协程的顺序阅读。

协程

yield生成器是特殊的迭代器；
greenlet 对 yield 进行了封装；
而 gevent 对 greenlet 进行了封装；
gevent 遇见延时操作就换任务执行，这里的延时操作可以是等待服务器资源或者sleep等等；

上面的概念会在后面的知识点进行讲解；

greenlet实现多任务

要使用greenlet，首先要安装greenlet

pip3 install greenlet

greenlet实现多任务代码

from greenlet import greenlet  import time    def task1():      while 1:          print("---1---")          gr2.switch()          time.sleep(1)    def task2():      while 1:          print("---2---")          gr1.switch()          time.sleep(1)      gr1 = greenlet(task1)  gr2 = greenlet(task2)  # 切换到gr1中执行  gr1.switch()

greenlet实现多任务

但注意，这里其实是一个单线程；并且经过测试，这里最后几句不能使用 __main__ ，否则会报错；

gevent实现多任务

可以看到，greenlet已经可以实现协程了，但需要我们手动进行任务切换，这样会很麻烦，因此我们要学习gevent，在greenlet的基础上进行了封装，可以帮助我们实现自动切换任务；

要使用gevent，使用要进行安装

pip3 install gevent

gevent实现多任务代码

import time  import gevent    def test1(n):      for i in range(n):          print("---test1---", gevent.getcurrent(), i)          # time.sleep(0.5)  # 这里使用time的sleep并不会因为耗时导致切换任务          gevent.sleep(0.5)      def test2(n):      for i in range(n):          print("---test2---", gevent.getcurrent(), i)          # time.sleep(0.5)  # 这里使用time的sleep并不会因为耗时导致切换任务          gevent.sleep(0.5)      def test3(n):      for i in range(n):          print("---test3---", gevent.getcurrent(), i)          # time.sleep(0.5)  # 这里使用time的sleep并不会因为耗时导致切换任务          gevent.sleep(0.5)      g1 = gevent.spawn(test1, 5)  g2 = gevent.spawn(test2, 5)  g3 = gevent.spawn(test3, 5)    g1.join()  g2.join()  g3.join()

gevent实现多任务.py

运行结果：

---test1--- <Greenlet at 0x1e9e64c2598: test1(5)> 0  ---test2--- <Greenlet at 0x1e9e64c26a8: test2(5)> 0  ---test3--- <Greenlet at 0x1e9e64c27b8: test3(5)> 0  ---test1--- <Greenlet at 0x1e9e64c2598: test1(5)> 1  ---test2--- <Greenlet at 0x1e9e64c26a8: test2(5)> 1  ---test3--- <Greenlet at 0x1e9e64c27b8: test3(5)> 1  ---test1--- <Greenlet at 0x1e9e64c2598: test1(5)> 2  ---test2--- <Greenlet at 0x1e9e64c26a8: test2(5)> 2  ---test3--- <Greenlet at 0x1e9e64c27b8: test3(5)> 2  ---test1--- <Greenlet at 0x1e9e64c2598: test1(5)> 3  ---test2--- <Greenlet at 0x1e9e64c26a8: test2(5)> 3  ---test3--- <Greenlet at 0x1e9e64c27b8: test3(5)> 3  ---test1--- <Greenlet at 0x1e9e64c2598: test1(5)> 4  ---test2--- <Greenlet at 0x1e9e64c26a8: test2(5)> 4  ---test3--- <Greenlet at 0x1e9e64c27b8: test3(5)> 4

运行结果

注意，在gevent中如果要使用sleep()，必须要使用 gevent.sleep()；

存在一个问题当我们创建g1，g2，g3时，如果不小心全部创建了g1，结果和没写错几乎是一样的；

问题版运行结果

g1 = gevent.spawn(test1, 5)  g2 = gevent.spawn(test2, 5)  g3 = gevent.spawn(test3, 5)    g1.join()  g1.join()  g1.join()    ---test1--- <Greenlet at 0x17d8ef12598: test1(5)> 0  ---test2--- <Greenlet at 0x17d8ef126a8: test2(5)> 0  ---test3--- <Greenlet at 0x17d8ef127b8: test3(5)> 0  ---test1--- <Greenlet at 0x17d8ef12598: test1(5)> 1  ---test2--- <Greenlet at 0x17d8ef126a8: test2(5)> 1  ---test3--- <Greenlet at 0x17d8ef127b8: test3(5)> 1  ---test1--- <Greenlet at 0x17d8ef12598: test1(5)> 2  ---test2--- <Greenlet at 0x17d8ef126a8: test2(5)> 2  ---test3--- <Greenlet at 0x17d8ef127b8: test3(5)> 2  ---test1--- <Greenlet at 0x17d8ef12598: test1(5)> 3  ---test2--- <Greenlet at 0x17d8ef126a8: test2(5)> 3  ---test3--- <Greenlet at 0x17d8ef127b8: test3(5)> 3  ---test1--- <Greenlet at 0x17d8ef12598: test1(5)> 4  ---test2--- <Greenlet at 0x17d8ef126a8: test2(5)> 4  ---test3--- <Greenlet at 0x17d8ef127b8: test3(5)> 4

问题版运行结果

协程的核心在于利用延时操作去做其他的任务；

给gevent打补丁

当我们使用gevent的时候，如果要延时操作，比如等待网络资源或者time.sleep()，必须要使用 gevent.sleep()，即每处延时操作都需要改成gevent的延时；如果我们想，还是按照原来的写法，并且使用gevent，怎么实现呢？这个实收，我们解疑使用打补丁的方法。只需要给使用gevent的代码添加如下一行代码即可完成打补丁

from gevent import monkey    monkey.patch_all()

使用打补丁的方式完成协程的使用

import time  import gevent  from gevent import monkey    monkey.patch_all()  def test1(n):      for i in range(n):          print("---test1---", gevent.getcurrent(), i)          time.sleep(0.5)  # 在打补丁的情况下等效于 gevent.sleep(0.5)    def test2(n):      for i in range(n):          print("---test2---", gevent.getcurrent(), i)          time.sleep(0.5)    def test3(n):      for i in range(n):          print("---test3---", gevent.getcurrent(), i)          time.sleep(0.5)    g1 = gevent.spawn(test1, 5)  g2 = gevent.spawn(test2, 5)  g3 = gevent.spawn(test3, 5)    g1.join()  g2.join()  g3.join()

给gevent打补丁.py

给gevent打补丁，使time.sleep(1)之类的耗时操作等效于gevent.sleep(1);

gevent.joinall()的使用

如果我们有很多函数要调用，那么岂不是得每次都先创建，在join()，gevent提供了一种简便方式；

import time  import gevent  from gevent import monkey    monkey.patch_all()  def test1(n):      for i in range(n):          print("---test1---", gevent.getcurrent(), i)          time.sleep(0.5)  # 在打补丁的情况下等效于 gevent.sleep(0.5)    def test2(n):      for i in range(n):          print("---test2---", gevent.getcurrent(), i)          time.sleep(0.5)    def test3(n):      for i in range(n):          print("---test3---", gevent.getcurrent(), i)          time.sleep(0.5)    gevent.joinall([      gevent.spawn(test1, 5),  # 括号内前面的是函数名，后面的是传参      gevent.spawn(test2, 5),      gevent.spawn(test3, 5),      ])

gevent.joinall()的使用.py

协程使用小案例-图片下载器

import urllib.request  import gevent  from gevent import monkey    monkey.patch_all()    def img_download(img_name, img_url):      req = urllib.request.urlopen(img_url)      data = req.read()      with open("images/"+img_name, "wb") as f:          f.write(data)      def main():      gevent.joinall([          gevent.spawn(img_download, "1.jpg", "https://rpic.douyucdn.cn/live-cover/appCovers/2019/05/13/6940298_20190513113912_small.jpg"),          gevent.spawn(img_download, "2.jpg", "https://rpic.douyucdn.cn/asrpic/190513/2077143_6233919_0d516_2_1818.jpg"),          gevent.spawn(img_download, "3.jpg", "https://rpic.douyucdn.cn/live-cover/appCovers/2018/11/24/1771605_20181124143723_small.jpg")      ])      if __name__ == "__main__":      main()

协程的使用-图片下载器.py

进程，线程，线程对比

区别

进程是资源分配的单位
线程是操作系统调度的单位
进程切换需要的资源很最大，效率很低
线程切换需要的资源一般，效率一般（当然了在不考虑GIL的情况下）
协程切换任务资源很小，效率高
多进程、多线程根据cpu核数不一样可能是并行的，但是协程是在一个线程中所以是并发。
多进程耗费的资源最多；
当我们python3运行一个py文件时，就是运行一个进程，进程中有一个默认的线程就是主线程，主线程拿着代码去执行；即进程是资源分配的单位，而线程才是真正拿着资源去执行，操作系统真正调度的就是线程；
一个进程里面有两个线程就是我们说的多线程的多任务方式，第二种多任务方式是多进程中有多线程；
线程的一大特点是可以利用某个线程在等待某个资源到来的时间去执行其他的任务；
在不考虑GIL的情况下，优先考虑协程，再考虑线程，再考虑进程；
进程是最稳定的，一个进程出问题了不会影响其他的进程，但耗费的资源较大；线程在切换任务时耗费的资源较线程少；协程可以利用线程在等待的时间做其他的事；

迭代器

迭代是访问集合元素的一种方式。迭代器是一个可以记住遍历的位置的对象。迭代器对象从集合的第一个元素开始访问，直到所有的元素被访问完结束。迭代器只能往前不会后退。

要理解协程的使用，首先要了解生成器；
要了解生成器，首先要理解迭代器；

推荐原来看过的一篇博客：一文彻底搞懂Python可迭代(Iterable)、迭代器(Iterator)和生成器(Generator)的概念，不过和本文关系不大，哈哈~

在了解迭代器之前，我们来认识两个单词

Iterable 可迭代的/可迭代/可迭代对象  Iterator 迭代器

可迭代

迭代器引入-for循环

In [1]: for i in [11,22,33]:     ...:     print(i)  11  22  33    In [2]: for i in "hhh":     ...:     print(i)  h  h  h    In [3]: for i in 10:     ...:     print(i)     ...:  ---------------------------------------------------------------------------  TypeError                                 Traceback (most recent call last)  <ipython-input-3-309758a01ba4> in <module>()  ----> 1 for i in 10:        2     print(i)        3  TypeError: 'int' object is not iterable  # “int”对象不可迭代

使用for循环时，in后面的数据类型是可迭代的才可以使用for循环，例如元组，列表，字符串等；不可迭代的，例如数字，小数点的；

判断是否可迭代

判断某个东西是否可迭代的，可以通过判断该数据类型是否为 Iterable 的子类，如果是则为可迭代；
isinstance 可以用来判断某对象是否是某类创建出来的；
比如我们要判断 a是否为A类创建出来的，可以使用 isinstance(a, A)进行判断；返回值为True，代表可迭代；

判断列表是否是可迭代的：

from collections import Iterable  isinstance([11,22,33], Iterable)  True

isinstance判断数据类型是否可迭代

In [6]: from collections import Iterable    In [7]: isinstance([11,22], Iterable)  Out[7]: True    In [8]: isinstance((11,22), Iterable)  Out[8]: True    In [9]: isinstance(10, Iterable)  Out[9]: False

元组，列表，字符串都是可迭代的；数字，小数不可迭代；

我们把可以通过for…in…这类语句迭代读取一条数据供我们使用的对象称之为可迭代对象（Iterable）。

自己定义的一个类，判断能不能用for？

自己创建一个类，满足能用for循环遍历的需求

不可迭代

class Classmate(object):      """docstring for Classmate"""      def __init__(self):          self.names = list()        def add(self, name):          self.names.append(name)      classmate = Classmate()    classmate.add("张三")  classmate.add("李四")  classmate.add("王五")    for name in classmate:      print(name)    # TypeError: 'Classmate' object is not iterable

可迭代对象本质

我们分析对可迭代对象进行迭代使用的过程，发现每迭代一次（即在for…in…中每循环一次）都会返回对象中的下一条数据，一直向后读取数据直到迭代了所有数据后结束。那么，在这个过程中就应该有一个“人”去记录每次访问到了第几条数据，以便每次迭代都可以返回下一条数据。我们把这个能帮助我们进行数据迭代的“人”称为迭代器(Iterator)。

可迭代对象的本质就是可以向我们提供一个这样的中间“人”即迭代器帮助我们对其进行迭代遍历使用。

可迭代对象通过__iter__方法向我们提供一个迭代器，我们在迭代一个可迭代对象的时候，实际上就是先获取该对象提供的一个迭代器，然后通过这个迭代器来依次获取对象中的每一个数据.

那么也就是说，一个具备了__iter__方法的对象，就是一个可迭代对象。

如果你不理解上面的话，没关系，你只需要知道 “如果想要将自己定义的一个类变为可迭代的，那么只需要在这个类中定义一个 __iter__ 方法即可”。

添加__iter__方法

class Classmate(object):      """docstring for Classmate"""      def __init__(self):          self.names = list()        def add(self, name):          self.names.append(name)        def __iter__(self):          pass    classmate = Classmate()    classmate.add("张三")  classmate.add("李四")  classmate.add("王五")    for name in classmate:      print(name)    # TypeError: iter() returned non-iterator of type 'NoneType'    # iter()返回“NoneType”类型的非迭代器

注意，这个时候的classmate已经是可迭代对象了，可以用isinstance(classmate, Iterable)验证;

但如果将__iter__()方法注释掉，就不是可迭代对象了，所以可以验证，要成为可迭代对象的第一步是添加__iter__()方法；

可迭代与迭代器

可迭代与迭代器

一个对象中有 __iter__ 方法，叫做可以迭代；
如果一个对象中有 __iter__ 方法，并且 __iter__ 方法返回一个另一个对象的引用，而返回的对象中又包含 __iter__ 和 __next__ 方法，那么这个返回的对象叫做迭代器；
只要有了迭代器，那么for方法就会通过迭代器中的 __next__ 方法来取值，每 for 循环一次，就调用一次 __next__ 方法；
使用 iter(xxxobj) 会自动调用 xxxobj 中的 __iter__ 方法，__iter__ 方法返回一个迭代器；
next(可迭代实例对象即 __iter__ 方法返回一个迭代器)，会自动去迭代器中调用 __next__ 方法；

一个可迭代的不一定是个迭代器；
一个迭代器一定可迭代；
(可迭代–里面有__iter__方法，迭代器–里面有__iter__和__next__方法)；

判断是否可迭代

以下列代码为例

for i in classmate

流程：

1.判断 classmate是否为可迭代的，即是否包含 __iter__ 方法；
2.如果第一步是可迭代的，那么就调用 iter(classmate) 即去调用 classmate 类中的 __iter__ 方法，返回一个迭代器，取返回值；
3.每 for 循环一次就去调用返回值中的 __next__ 方法一次，__next__ 返回什么，就给i什么；

自定义使用for循环步骤

1.在类中添加 __iter__ 方法；
2.__iter__ 方法返回一个对象的引用，这个对象必须包含 __iter__ 和 __next__ 方法；
3.在包含 __iter__ 和 __next__ 方法的类中，编写 __next__ 方法返回值；

for…in…循环的本质

for item in Iterable

循环的本质就是先通过iter()函数获取可迭代对象Iterable的迭代器，然后对获取到的迭代器不断调用next()方法来获取下一个值并将其赋值给item，当遇到StopIteration的异常后循环结束。

完善自定义迭代器

一个实现了__iter__方法和__next__方法的对象，就是迭代器。

让迭代器可以完整返回所有的数据；

import time  from collections.abc import Iterable, Iterator      class Classmate(object):      def __init__(self):          self.names = list()        def add(self, name):          self.names.append(name)        def __iter__(self):          return ClassmateIterable(self)      class ClassmateIterable(object):      def __init__(self, obj):          self.obj = obj          self.num = 0        def __iter__(self):          pass        def __next__(self):          # return self.obj.names[0]          try:              ret = self.obj.names[self.num]              self.num += 1              return ret          except IndexError as e:              raise StopIteration      def main():      classmate = Classmate()      classmate.add("张三")      classmate.add("李四")      classmate.add("王五")      print("判断classmate是否为可迭代的：", isinstance(classmate, Iterable))      classmate_iterator = iter(classmate)      print("判断classmate_iterator是否为迭代器：", isinstance(classmate_iterator, Iterator))      # 调用一次 __next__      print("classmate_iterator's next:", next(classmate_iterator))      for i in classmate:          print(i)          time.sleep(1)      if __name__ == '__main__':      main()

自己实现一个可迭代的对象

可以看到，现在已经可以实现for循环使用自定义的类了；但在这个代码里我们看到为了实现返回迭代器我们要再定义一个额外的类，这样是比较麻烦的。在这里我们可以进行简化一下，不返回另一个类，而是返回自己这个类，并且在自己类中定义一个 __next__ 方法。简化如下

改进简化迭代器

import time  from collections.abc import Iterable, Iterator      class Classmate(object):      def __init__(self):          self.names = list()          self.num = 0        def add(self, name):          self.names.append(name)        def __iter__(self):          return self        def __next__(self):          # return self.obj.names[0]          try:              ret = self.names[self.num]              self.num += 1              return ret          except IndexError as e:              raise StopIteration      def main():      classmate = Classmate()      classmate.add("张三")      classmate.add("李四")      classmate.add("王五")      for i in classmate:          print(i)          time.sleep(1)      if __name__ == '__main__':      main()

改进简化迭代器.py

迭代器的应用

迭代器的作用

不用迭代器，是当要做某事之前，就生成并存储数据，存储数据时可能会占用大量的空间；
用迭代器，是掌握数据的生成方法，什么时候使用，什么时候生成；
比如range(10)，即时生成10个数据，那么range(1000000000)呢？
range：生成10个值的列表；xrange：存储生成10个值的方式；
python2 中 range(10) 存储的是一个列表，xrange(10) 存储的是生成10个值的方式，是一个迭代器；
python3 中 range() 已经相当于python2中的 xrange()了，并且py3中已经没有xrange()了；
迭代器是存储生成数据的方式，而不是存储数据结果；

python3中使用range：

>>> range(10)  range(0, 10)  >>> ret = range(10)  >>> next(ret)  Traceback (most recent call last):    File "<pyshell#3>", line 1, in <module>      next(ret)  TypeError: 'range' object is not an iterator  >>> for i in range(10):      print(i)  0  1  2  3  ...

正常实现斐波那契数列

nums = []    a = 0  b = 1  i = 0  while i < 10:      nums.append(a)      a, b = b, a+b      i += 1    for i in nums:      print(i)

使用迭代器实现斐波那契数列

class Fibonacci(object):      def __init__(self, times):          self.times = times          self.a = 0          self.b = 1          self.current_num = 0        def __iter__(self):          return self        def __next__(self):          if self.current_num < self.times:              ret = self.a              self.a, self.b = self.b, self.a+self.b              self.current_num += 1              return ret          else:              raise StopIteration      fibo = Fibonacci(10)    for i in fibo:      print(i)

使用迭代器实现斐波那契数列

什么时候调，什么时候生成。

迭代器使用的其他方式-列表元组等类型转换

当我们使用 list() 或者 tuple() 进行类型转换时，使用的其实也是迭代器；

a = (11,22,33)  b = list(a)

当我们使用list()将元组转换成列表时，是使用了迭代器的原理，先定义一个空列表，用迭代器通过 __next__ 从元组中取第一个值，添加到空列表中，再依次从元组取值，添加入列表，直到元组中没有值了，主动抛出迭代停止异常；
同理，将列表转换成元组也是如此；

生成器

迭代器：用来节省内存空间而且还知道将来怎么生成数据的方式；
生成器：一种特殊的迭代器；

生成器方式：

1.将列表推导式的小括号换成中括号；
2.函数中使用yield

实现生成器方式1

In [15]: L = [ x*2 for x in range(5)]    In [16]: L  Out[16]: [0, 2, 4, 6, 8]    In [17]: G = ( x*2 for x in range(5))    In [18]: G  Out[18]: <generator object <genexpr> at 0x7f626c132db0>    In [19]: next(G)  Out[19]: 0    In [20]: next(G)  Out[20]: 2

实现生成器方式2

使用yield的生成器

def Fibonacci(n):      a, b = 0, 1      count_num = 0      while count_num < n:          # 如果函数中有一个yield语句，那么这个就不再是函数，而是一个生成器的模板          yield a          a, b = b, a+b          count_num += 1    # 如果在调用时发现这个函数中有yield，那么此时，不是调用函数，而是创建一个生成器对象  fb = Fibonacci(5)    print("使用for循环遍历生成器中的所有数字".center(40, "-"))  for i in fb:      print(i)

使用yield完成斐波那契数列

生成器执行流程：当第一次调用for/next执行时，会从生成器的第一行开始依次向下执行，直到在循环中碰见yield，就会返回yield后面的变量/字符；然后第二次调用for/next时，就会从上次的yield后面的代码继续执行，直到在循环中再次碰到yield，返回；依次往下，直到没有了数据。

可以使用 for i in 生成器对象来遍历生成器中的数据，也可以用 next(生成器对象) 来一个一个获取生成器中的值；

使用next获取生成器中的值

def Fibonacci(n):      a, b = 0, 1      count_num = 0      while count_num < n:          # 如果函数中有一个yield语句，那么这个就不再是函数，而是一个生成器的模板          yield a          a, b = b, a+b          count_num += 1    # 如果在调用时发现这个函数中有yield，那么此时，不是调用函数，而是创建一个生成器对象  fb = Fibonacci(5)    print("使用next依次生成三次数字".center(40, "-"))  print(next(fb))  print(next(fb))  print(next(fb))    print("使用for循环遍历剩余的数字".center(40, "-"))  for i in fb:      print(i)

使用next获取生成器中的值

生成器-send方式

可以重复创建多个生成器，多个生成器之间互不干扰；
如果在生成器中有return值，可以在生成器结束后用出错的结果.value 来进行接收；

def Fibonacci(n):      a, b = 0, 1      count_num = 0      while count_num < n:          # 如果函数中有一个yield语句，那么这个就不再是函数，而是一个生成器的模板          yield a          a, b = b, a+b          count_num += 1      return "okhaha"    # 如果在调用时发现这个函数中有yield，那么此时，不是调用函数，而是创建一个生成器对象  fb = Fibonacci(5)    while 1:      try:          result = next(fb)          print(result)        except Exception as e:          print(e.value)          break

生成器使用send

除了使用next来启动生成器之外，还可以使用send来启动生成器；

def Fibonacci(n):      a, b = 0, 1      count_num = 0      while count_num < n:          ret = yield a          print("ret:", ret)          a, b = b, a+b          count_num += 1    fb = Fibonacci(5)    print(next(fb))  print(fb.send("haha"))  print(next(fb))    # 0  # ret: haha  # 1  # ret: None  # 1

使用send来启动生成器

我们可以理解为，第一次使用next，先执行等号右边的代码，就将yield a返回给了next(fb)；然后下次调用send时，执行等号左边的，将send的传值赋值给ret，再执行后续代码；
或者我们可以理解 ret = yield a 为两步 ===>1.yield a； 2.ret = arg；其中的arg表示send的传值，如果不传值，默认为None，所以当next在send后面调用时，就默认传了None；

注意，一般不将send用作第一次唤醒生成器，如果一定要使用send第一次唤醒，要send(None);

生成器-小总结
生成器特点：

一个没有__iter__和__next__方法的特殊迭代器；
函数只执行一部分就返回；
可以让一个函数暂停执行，并且保存上次的值，根据上次的值恢复到原来的样子，再做接下来的操作；
迭代器节省空间，实现循环；
生成器可以让一个看起来像函数的代码暂停执行，并根据自己的想法调用next/send继续执行；

使用yield完成多任务

在python2中，while1 的执行时间大概是while True的2/3，这是因为True在2中不是关键字，可以随意赋值，因此用while 1；
在python3中，True已经是关键字了，解释器不用判断True的值，所以while True和 while 1的区别不大，但可能还是1更快；

进程之间切换任务，占用的资源很大，创建进程，释放进程需要浪费大量的时间，进程的效率没有线程高，比线程占用资源更少的是协程；

使用yield完成多任务

import time      def task_1():      while 1:          print("---1---")          time.sleep(0.5)          yield      def task_2():      while 1:          print("---2---")          time.sleep(0.5)          yield      def main():      t1 = task_1()      t2 = task_2()      while 1:          next(t1)          next(t2)      if __name__ == "__main__":      main()

使用yield完成多任务

是假的多任务，属于并发；