协程:
是单线程下的并发,又称微线程,纤程。英文名Coroutine。一句话说明什么是线程:协程是一种用户态的轻量级线程,即协程是由用户程序自己控制调度的。
#1. python的线程属于内核级别的,即由操作系统控制调度(如单线程遇到io或执行时间过长就会被迫交出cpu执行权限,切换其他线程运行)
#2. 单线程内开启协程,一旦遇到io,就会从应用程序级别(而非操作系统)控制切换,以此来提升效率(!!!非io操作的切换与效率无关)
协程拥有自己的寄存器上下文和栈。协程调度切换时,将寄存器上下文和栈保存到其他地方,在切回来的时候,恢复先前保存的寄存器上下文和栈
协程能实现并发。
优点:
- 无需线程上下文切换的开销
- 无需考虑数据安全问题
- 方便切换控制留,简化编程模型,单线程内就可以实现并发的效果,最大限度地利用cpu
- 高并发,高扩展性,低成本
缺点:
- 协程的本质是单线程,无法利用多核资源,不能实现并行,可以是一个程序开启多个进程,每个进程内开启多个线程,每个线程内开启协程,因而一旦协程出现阻塞,将会阻塞整个线程
- 只适用I/O操作密集型,一个协程遇到IO操作自动切换到其他协程
Greenletmo模块实现协程:
如果我们在单个线程内有20个任务,要想实现在多个任务之间切换,而使用greenlet模块可以非常简单地实现这20个任务直接的切换
from greenlet import greenlet import time def eat(name): print('%s eat 1' %name) time.sleep(3) g2.switch('alex') print('%s eat 2' %name) g2.switch() def play(name): print('%s play 1' %name) g1.switch() print('%s play 2' %name) g1=greenlet(eat) g2=greenlet(play) if __name__ == '__main__': g1.switch('egon') #可以在第一次switch时传入参数,以后都不需要
单纯的切换(在没有io的情况下或者没有重复开辟内存空间的操作),反而会降低程序的执行速度
#顺序执行 import time def f1(): res=1 for i in range(100000000): res+=i def f2(): res=1 for i in range(100000000): res*=i start=time.time() f1() f2() stop=time.time() print('run time is %s' %(stop-start)) #10.985628366470337 #切换 from greenlet import greenlet import time def f1(): res=1 for i in range(100000000): res+=i g2.switch() def f2(): res=1 for i in range(100000000): res*=i g1.switch() start=time.time() g1=greenlet(f1) g2=greenlet(f2) g1.switch() stop=time.time() print('run time is %s' %(stop-start)) # 52.763017892837524
Gevent模块实现协程:
Python通过yield提供了对协程的基本支持,但是不完全。而第三方的gevent为Python提供了比较完善的协程支持。
gevent是第三方库,通过greenlet实现协程,其基本思想是:
当一个greenlet遇到IO操作时,比如访问网络,就自动切换到其他的greenlet,等到IO操作完成,再在适当的时候切换回来继续执行。由于IO操作非常耗时,经常使程序处于等待状态,有了gevent为我们自动切换协程,就保证总有greenlet在运行,而不是等待IO。
由于切换是在IO操作时自动完成,所以gevent需要修改Python自带的一些标准库,这一过程在启动时通过monkey patch完成:
#用法
g1=gevent.spawn(func,1,,2,3,x=4,y=5)创建一个协程对象g1,spawn括号内第一个参数是函数名,如eat,后面可以有多个参数,可以是位置实参或关键字实参,都是传给函数eat的
g2=gevent.spawn(func2)
g1.join() #等待g1结束
g2.join() #等待g2结束
#或者上述两步合作一步:gevent.joinall([g1,g2])
g1.value#拿到func1的返回值
遇到IO阻塞时会自动切换任务
import gevent #1.检测IO #2.自动切换 def eat(name): print('%s eat 1' %name) gevent.sleep(2) print('%s eat 2' %name) def play(name): print('%s play 1' %name) gevent.sleep(1) print('%s play 2' %name) start=time.time() g1=gevent.spawn(eat,'alex') g2=gevent.spawn(play,'egon') # g1.join() # g2.join() gevent.joinall([g1,g2]) stop=time.time() print(stop-start)
上例gevent.sleep(2)模拟的是gevent可以识别的io阻塞,
而time.sleep(2)或其他的阻塞,gevent是不能直接识别的需要用下面一行代码,打补丁,就可以识别了
from gevent import monkey;monkey.patch_all()必须放到被打补丁者的前面,如time,socket模块之前
或者我们干脆记忆成:要用gevent,需要将from gevent import monkey;monkey.patch_all()放到文件的开头
#协程:单线程下实现并发,用户从应用程序级别控制单线程下任务的切换,注意一定是遇到IO才切 import gevent from threading import current_thread import time def eat(): print('%s eat 1' %current_thread().getName()) gevent.sleep(2) print('%s eat 2' %current_thread().getName()) def play(): print('%s play 1' % current_thread().getName()) gevent.sleep(2) print('%s play 2' % current_thread().getName()) start=time.time() g1=gevent.spawn(eat,) g2=gevent.spawn(play,) #g1.join() #g2.join() gevent.joinall([g1,g2]) stop=time.time() print(stop-start) ################################################# from gevent import monkey;monkey.patch_all() # gevent识别不了别人的IO,要想实现得先声明 import gevent,os from threading import current_thread #1.检测IO #2.自动切换 import time def eat(): print('%s eat 1' %current_thread().getName())# os.getpid() time.sleep(2) print('%s eat 2' %current_thread().getName()) def play(): print('%s play 1' %current_thread().getName()) time.sleep(1) print('%s play 2' %current_thread().getName()) start=time.time() g1=gevent.spawn(eat,) g2=gevent.spawn(play,) # g1.join() # g2.join() gevent.joinall([g1,g2]) stop=time.time() print(stop-start)
我们可以用threading.current_thread().getName()来查看每个g1和g2,查看的结果为DummyThread-n,即假线程
单线程并发爬页面
from gevent import monkey;monkey.patch_all() import gevent import requests from threading import current_thread def get(url): print('%s get %s' %(current_thread().getName(),url)) response=requests.get(url) if response.status_code == 200: return {'url':len(response.text)} # print({'url':len(response.text)}) g1=gevent.spawn(get,'http://www.baidu.com') g2=gevent.spawn(get,'http://www.python.org') g3=gevent.spawn(get,'http://www.jd.com') g1.join() g2.join() g3.join() print(g1.value) print(g2.value) print(g3.value)
yield与协程
import time """ 传统的生产者-消费者模型是一个线程写消息,一个线程取消息,通过锁机制控制队列和等待,但一不小心就可能死锁。 如果改用协程,生产者生产消息后,直接通过yield跳转到消费者开始执行,待消费者执行完毕后,切换回生产者继续生产,效率极高。 """ # 注意到consumer函数是一个generator(生成器): # 任何包含yield关键字的函数都会自动成为生成器(generator)对象 def consumer(): r = '' while True: # 3、consumer通过yield拿到消息,处理,又通过yield把结果传回; # yield指令具有return关键字的作用。然后函数的堆栈会自动冻结(freeze)在这一行。 # 当函数调用者的下一次利用next()或generator.send()或for-in来再次调用该函数时, # 就会从yield代码的下一行开始,继续执行,再返回下一次迭代结果。通过这种方式,迭代器可以实现无限序列和惰性求值。 n = yield r if not n: return print('[CONSUMER] ←← Consuming %s...' % n) time.sleep(1) r = '200 OK' def produce(c): # 1、首先调用c.next()启动生成器 next(c) n = 0 while n < 5: n = n + 1 print('[PRODUCER] →→ Producing %s...' % n) # 2、然后,一旦生产了东西,通过c.send(n)切换到consumer执行; cr = c.send(n) # 4、produce拿到consumer处理的结果,继续生产下一条消息; print('[PRODUCER] Consumer return: %s' % cr) # 5、produce决定不生产了,通过c.close()关闭consumer,整个过程结束。 c.close() if __name__=='__main__': # 6、整个流程无锁,由一个线程执行,produce和consumer协作完成任务,所以称为“协程”,而非线程的抢占式多任务。 c = consumer() produce(c) ''' result: [PRODUCER] →→ Producing 1... [CONSUMER] ←← Consuming 1... [PRODUCER] Consumer return: 200 OK [PRODUCER] →→ Producing 2... [CONSUMER] ←← Consuming 2... [PRODUCER] Consumer return: 200 OK [PRODUCER] →→ Producing 3... [CONSUMER] ←← Consuming 3... [PRODUCER] Consumer return: 200 OK [PRODUCER] →→ Producing 4... [CONSUMER] ←← Consuming 4... [PRODUCER] Consumer return: 200 OK [PRODUCER] →→ Producing 5... [CONSUMER] ←← Consuming 5... [PRODUCER] Consumer return: 200 OK '''