协程Coroutine的概念

协程,多线程,多进程,都是用来实现并发的.

由于现代操作系统的时间片原理,可以将这个概念抽象来看,只要能够满足保存上下文,在中断的地方回来执行的程序,都可以用来实现并发.所以并发的本质就是切换+保存状态,只要能够做到这两点,都可以实现并发.

在单线程内部,不可避免的会有IO操作,如果我们的程序能够编写的让单线程在出现IO操作的时候,转到另外的计算部分继续操作,对于操作系统来讲,我这一个线程就会一直处在不被IO阻塞而就绪进行计算的状态,这也是提高效率的一种办法.

协程的本质就是在单线程下，由用户自己控制一个任务遇到io阻塞了就切换另外一个任务去执行，以此来提升效率。为了实现它，我们需要找寻一种可以同时满足以下条件的解决方案： #1. 可以控制多个任务之间的切换，切换之前将任务的状态保存下来，以便重新运行时，可以基于暂停的位置继续执行。 #2. 作为1的补充：可以检测io操作，在遇到io操作的情况下才发生切换

之前学过生成器写的生产者消费者模型就是单线程内实现并发,让两个程序来回切换.如果函数用yield来接收参数然后处理,就类似一个协程.而且由于第一个yield无法接收,必须先用内置函数next()启动生成器,这就像多线程和多进程的.start()方法一样,启动一个协程.

# yield模拟协程,实现一个查找功能的例子:
def find(pattern):
    while True:
        search = yield
        if pattern in search:
            print('found {} in {}'.format(pattern, search))
        else:
            print('Not searched')


searchlist = ['cony', 'momisjenny', 'testofcony', ""cony's summer"", 'minkolovejenny']
a = find('jenny')  # 传入要查找的内容
next(a)  # 启动协程
for i in searchlist:  # 协程等待接收传入的数据,传入数据如果包含要查找的内容就打印找到,否则就打印找不到.
    a.send(i)

可以发现,启动生成器之后,生成器像一个真的线程一样一直在提供服务;主程序不需要使用这个功能模块的时候,find函数完全不起作用,需要用到的时候,只要传入查找内容就可以返回结果,就像去另外一个线程或者进程请求结果一样.但实际上,还是在一个单线程内执行,如果给协程里边加上IO操作(比如添加一个input语句),协程会阻塞,这时候整个线程依然会阻塞,并不像真正的协程,究其原因还是因为yield是无法在自身发生阻塞的情况下跳出的.这就需要用一些模块来帮助在单线程内遇到IO的时候切换执行内容.

greenlet模块

greenlet 用于在程序之间切换

from greenlet import greenlet
import time


def eat():
    print('start eating')
    time.sleep(1)
    g2.switch()
    print('end eating')
    g2.switch()


def play():
    print('start playing')
    time.sleep(1)
    g1.switch()
    print('end playing')
    

g1 = greenlet(eat)
g2 = greenlet(play)
g1.switch()

将两个函数都交给greenlet去代理,得到两个对象g1和g2,之后先用一个switch()方法启动g1,就会从头执行eat(),然后执行到g2.switch()的时候会保存一个断点,再到g2执行,第一次跳入g2的时候也是从头开始执行play(),遇到跳回g1的时候,保存这里的断点,然后返回g1上次的断点继续往下执行,又遇到g2.switch(),返回g2的断点,继续执行.这里的例子比较简单,如果用在循环内,可以实现两个程序来回切换.

上边的例子如果用g2.switch()启动,结果就会变成start playing --> start eating --> end playing.

不过这个时候发现,如果在函数内加上IO,依然会阻塞,说明greenlet只是提供了一个顺序流程的改变,还不是真正的协程效果.需要一种可以检测IO状态的模块.gevent模块就是在greenlet的基础上加上了检测IO的功能,实现真正的协程.

gevent模块

import gevent

def eat(name):
    print(name,'start eating')
    gevent.sleep(2)  # 这里用gevent的sleep方法,可以被gevent感知到
    print(name,'end eating')
    return 12

def play(name):
    print(name,'start playing')
    gevent.sleep(2)
    print(name,'end playing')
    return 21


g1=gevent.spawn(eat,name='cony')  # 把代码注册到一个协程里,只用spawn的话无法运行,必须使用join方法才行
g2=gevent.spawn(play,name='jenny')  # 又注册了一个协程
g1.join()  
g2.join()  # 也可以用gevent.joinall([g1,g2])
print(g1.value)  # 取得返回值12
print(g2.value)  # 取得返回值21

由于协程一开始可以异步提交,但执行到最后,肯定要和当前线程内程序最后同步并获得结果,否则就失去了协程的意义.这个模块不能够感知time.sleep(),但是可以感知自己的sleep()方法,在发现sleep了之后,就会切换执行其他gevent对象内的程序.time.sleep(2)或其他的阻塞,gevent是不能直接识别的需要用到下边代码from gevent import monkey;monkey.patch_all()(写在一行内),放到要导入的模块之前,最好放在文件开头的地方就行了.修改后代码如下:

from gevent import monkey
import gevent
import time

monkey.patch_all()


def eat(name):
    print(name, 'start eating')
    time.sleep(2)
    print(name, 'end eating')
    return 10


def play(name):
    print(name, 'start playing')
    time.sleep(2)
    print(name, 'end playing')
    return 11


g1 = gevent.spawn(eat, name='cony')
g2 = gevent.spawn(play, name='jenny')
g1.join()
g2.join()
print(g1.value)
print(g2.value)

协程模块遇到能够识别的IO操作的时候,才会进行任务切换,实现并发的效果.

线程是CPU调度的最小单位,所以协程实际上操作系统是不管的,是由用户自行调度的.协程本质上是在一个线程内部来回切换.有效利用协程可以提高单线程的效率.

gevent模块应用

# 模拟高IO场景
from gevent import monkey;monkey.patch_all()
import gevent
import time

def task(i):
    time.sleep(0.5)
    print(12345,i)

def sync():
    for i in range(10):
        task(i)

def async():
    t_list = [gevent.spawn(task,i) for i in range(10)]
    gevent.joinall(t_list)

sync()
async()

从执行结果来看,显然异步的效率高,协程适合单线程内IO操作频繁的时候,所以协程很适合用在网络IO方面的处理.

# 轻型爬虫
from gevent import monkey

monkey.patch_all()
import gevent
import requests


def get_page(url, filename):
    print('GET: %s' % url)
    reponse = requests.get(url)
    if reponse.status_code == 200:
        with open(filename, 'w') as f:
            f.write(reponse.content.decode('utf-8'))
        print('{}写入完成'.format(url))


url_list = ['http://www.conyli.cc',
            'http://www.baidu.com',
            'http://github.com',
            'http://bbs.saraba1st.com/2b/']

filename = 'html'
res_list = []
for i in range(len(url_list)):
    res_list.append(gevent.spawn(get_page, url_list[i], filename + str(i) + '.html'))
gevent.joinall(res_list)

爬取内容各个网址的内容写入到文件里.从结果里可以看到,写入完成并不是按照启动的顺序,而是先爬完的先读写.这样比同步执行各个操作要节省很多时间.

学到这里,可以看到,单线程内部可以有协程,多线程,还有多进程,因此如果有大规模的并行任务,理论上是可以开i个进程,每个进程中有j个线程,每个线程中有k个协程,也就是一共ijk数量的任务运行.一般在现代操作系统中,4核CPU的这个总数量可以支持50000个.

协程实现单线程的socketserver并发

# 思路和之前多线程或者多进程的方式类似,用一个循环不断监听socket对象的accpet()阻塞,每生成一个连接对象,就用handler方法注册到新的协程里去处理这个连接对象.
from gevent import monkey

monkey.patch_all()
import gevent
import socket

ip_port = ('127.0.0.1', 8080)
buffer_size = 1024
backlog = 5


def get_conn():
    while True:
        conn, addr = sk.accept()
        gevent.spawn(handler, conn, addr)


def handler(conn, addr):
    print('From {} connected'.format(addr))
    try:
        while True:
            data = conn.recv(buffer_size)
            # print('From {}: {}'.format(addr, data.decode('utf-8')))
            conn.send(data.upper())
    except Exception:
        print('From {} connect lost'.format(addr))
        conn.close()


sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sk.bind(ip_port)
sk.listen(backlog)
get_conn()

多进程,多线程,协程都实现过socketserver了,再回头看socketserver模块的源码,其实就是用多线程Threading实现的.这里有篇文章提到了说一般可以考虑多进程+协程的方式工作,又可以跑满CPU,又不用考虑多线程加锁问题.另外,gevent还有协程池pool等方法,留待以后再来研究了.