一、协程介绍
协程:英文名Coroutine,是单线程下的并发,又称微线程,纤程。
协程是一种用户态的轻量级线程,即协程是由用户程序自己控制调度的。对比操作系统控制线程的切换,用户在单线程内控制协程的切换。
协程自己本身无法实现并发(甚至性能会降低),协程+IO切换性能提高。
1、介绍
通常程序中子程序调用总是一个入口,一次返回,调用顺序是明确的。而协程的调用和子程序不同。
协程看上去也是子程序,但执行过程中,在子程序内部可中断,然后转而执行别的子程序,在适当的时候再返回来接着执行。
注意,在一个子程序中中断,去执行其他子程序,不是函数调用,有点类似CPU的中断。
看起来A、B的执行有点像多线程,但协程的特点在于是一个线程执行,那和多线程比,协程有何优势?
最大的优势就是协程极高的执行效率。因为子程序切换不是线程切换,而是由程序自身控制,因此,没有线程切换的开销,和多线程比,线程数量越多,协程的性能优势就越明显。
第二大优势就是不需要多线程的锁机制,因为只有一个线程,也不存在同时写变量冲突,在协程中控制共享资源不加锁,只需要判断状态就好了,所以执行效率比多线程高很多。
因为协程是一个线程执行,那怎么利用多核CPU呢?最简单的方法是多进程+协程,既充分利用多核,又充分发挥协程的高效率,可获得极高的性能。
2、举例
Python对协程的支持是通过generator实现的。
在generator中,我们不但可以通过for
循环来迭代,还可以不断调用next()
函数获取由yield
语句返回的下一个值。
但是Python的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
|
def consumer(): r = '' while True : n = yield r if not n: return print ( '[CONSUMER] Consuming %s...' % n) r = '200 OK' def produce(c): c.send( None ) n = 0 while n < 5 : n = n + 1 print ( '[PRODUCER] Producing %s...' % n) r = c.send(n) print ( '[PRODUCER] Consumer return: %s' % r) c.close() c = consumer() produce(c) |
执行结果:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
[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 |
注意到consumer
函数是一个generator
,把一个consumer
传入produce
后:
-
首先调用
c.send(None)
启动生成器; -
然后,一旦生产了东西,通过
c.send(n)
切换到consumer
执行; -
consumer
通过yield
拿到消息,处理,又通过yield
把结果传回; -
produce
拿到consumer
处理的结果,继续生产下一条消息; -
produce
决定不生产了,通过c.close()
关闭consumer
,整个过程结束。
整个流程无锁,由一个线程执行,produce
和consumer
协作完成任务,所以称为“协程”,而非线程的抢占式多任务。
最后套用Donald Knuth的一句话总结协程的特点:“子程序就是协程的一种特例。”
3、优点如下:
- 协程的切换开销更小,属于程序级别的切换,操作系统完全感知不到,因而更加轻量级
- 单线程内就可以实现并发的效果,最大限度地利用cpu
4、缺点如下:
- 协程的本质是单线程下,无法利用多核,可以是一个程序开启多个进程,每个进程内开启多个线程,每个线程内开启协程
- 协程指的是单个线程,因而一旦协程出现阻塞,将会阻塞整个线程
5、总结协程特点:
- 必须在只有一个单线程里实现并发
- 修改共享数据不需加锁
- 用户程序里自己保存多个控制流的上下文栈
- 附加:一个协程遇到IO操作自动切换到其它协程(如何实现检测IO,yield、greenlet都无法实现,就用到了gevent模块(select机制))
二、greenlet(绿叶)模块
如果我们在单个线程内有20个任务,要想实现在多个任务之间切换,使用yield生成器的方式过于麻烦(需要先得到初始化一次的生成器,然后再调用send。。。非常麻烦),而使用greenlet模块可以非常简单地实现这20个任务直接的切换。
1、安装模块
pip3 install greenlet
2、greenlet实现状态切换
单纯的切换(在没有io的情况下或者没有重复开辟内存空间的操作),反而会降低程序的执行速度。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
from greenlet import greenlet def eat(name): print ( '%s eat 1' % name) g2.switch( 'nick' ) 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) g1.switch( 'nick' ) # 可以在第一次switch时传入参数,以后都不需要 |
3、效率对比
greenlet只是提供了一种比generator更加便捷的切换方式,当切到一个任务执行时如果遇到io,那就原地阻塞,仍然是没有解决遇到IO自动切换来提升效率的问题。
单线程里的这20个任务的代码通常会既有计算操作又有阻塞操作,我们完全可以在执行任务1时遇到阻塞,就利用阻塞的时间去执行任务2...如此,才能提高效率,这就用到了Gevent模块。
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
|
#顺序执行 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模块
Gevent 是一个第三方库,可以轻松实现并发同步或异步编程,在gevent中用到的主要模式是Greenlet,它是以C扩展模块形式接入Python的轻量级协程。
1、安装
pip3 install gevent
2、 用法介绍
1
2
3
4
5
6
7
8
9
|
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的返回值 |
1、遇到io主动切换
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
import gevent 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) g1 = gevent.spawn(eat, 'egon' ) g2 = gevent.spawn(play,name = 'egon' ) g1.join() g2.join() # 或者gevent.joinall([g1,g2]) print ( '主' ) |
上例gevent.sleep(2)
模拟的是gevent可以识别的io阻塞,而time.sleep(2)
或其他的阻塞,gevent是不能直接识别的需要用下面一行代码,打补丁,就可以识别了。
1
|
from gevent import monkey;monkey.patch_all() |
必须放到被打补丁者的前面,如time,socket模块之前。或者我们干脆记忆成:要用gevent,需要将from gevent import monkey;monkey.patch_all()
放到文件的开头。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
from gevent import monkey;monkey.patch_all() import gevent import time def eat(): print ( 'eat food 1' ) time.sleep( 2 ) print ( 'eat food 2' ) def play(): print ( 'play 1' ) time.sleep( 1 ) print ( 'play 2' ) g1 = gevent.spawn(eat) g2 = gevent.spawn(play) gevent.joinall([g1,g2]) print ( '主' ) |
2、 查看threading.current_thread().getName()
我们可以用threading.current_thread().getName()
来查看每个g1和g2,查看的结果为DummyThread-n,即假线程
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
from gevent import monkey;monkey.patch_all() import threading import gevent import time def eat(): print (threading.current_thread().getName()) print ( 'eat food 1' ) time.sleep( 2 ) print ( 'eat food 2' ) def play(): print (threading.current_thread().getName()) print ( 'play 1' ) time.sleep( 1 ) print ( 'play 2' ) g1 = gevent.spawn(eat) g2 = gevent.spawn(play) gevent.joinall([g1,g2]) print ( '主' ) |
3、Gevent之同步与异步
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
|
from gevent import spawn,joinall,monkey;monkey.patch_all() import time def task(pid): """ Some non-deterministic task """ time.sleep( 0.5 ) print ( 'Task %s done' % pid) def synchronous(): # 同步 for i in range ( 10 ): task(i) def asynchronous(): # 异步 g_l = [spawn(task,i) for i in range ( 10 )] joinall(g_l) print ( 'DONE' ) if __name__ = = '__main__' : print ( 'Synchronous:' ) synchronous() print ( 'Asynchronous:' ) asynchronous() # 上面程序的重要部分是将task函数封装到Greenlet内部线程的gevent.spawn。 # 初始化的greenlet列表存放在数组threads中,此数组被传给gevent.joinall 函数, # 后者阻塞当前流程,并执行所有给定的greenlet任务。执行流程只会在 所有greenlet执行完后才会继续向下走。 |
4、Gevent之应用
通过gevent实现单线程下的socket并发
注意:from gevent import monkey;monkey.patch_all()一定要放到导入socket模块之前,否则gevent无法识别socket的阻塞。
1、 服务端
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
|
from gevent import monkey;monkey.patch_all() from socket import * import gevent #如果不想用money.patch_all()打补丁,可以用gevent自带的socket # from gevent import socket # s=socket.socket() def server(server_ip,port): s = socket(AF_INET,SOCK_STREAM) s.setsockopt(SOL_SOCKET,SO_REUSEADDR, 1 ) s.bind((server_ip,port)) s.listen( 5 ) while True : conn,addr = s.accept() gevent.spawn(talk,conn,addr) def talk(conn,addr): try : while True : res = conn.recv( 1024 ) print ( 'client %s:%s msg: %s' % (addr[ 0 ],addr[ 1 ],res)) conn.send(res.upper()) except Exception as e: print (e) finally : conn.close() if __name__ = = '__main__' : server( '127.0.0.1' , 8080 ) |
2、多线程并发多个客户端
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
from threading import Thread from socket import * import threading def client(server_ip,port): c = socket(AF_INET,SOCK_STREAM) #套接字对象一定要加到函数内,即局部名称空间内,放在函数外则被所有线程共享,则大家公用一个套接字对象,那么客户端端口永远一样了 c.connect((server_ip,port)) count = 0 while True : c.send(( '%s say hello %s' % (threading.current_thread().getName(),count)).encode( 'utf-8' )) msg = c.recv( 1024 ) print (msg.decode( 'utf-8' )) count + = 1 if __name__ = = '__main__' : for i in range ( 500 ): t = Thread(target = client,args = ( '127.0.0.1' , 8080 )) t.start() |
到此这篇关于Python协程操作模块的文章就介绍到这了。希望对大家的学习有所帮助,也希望大家多多支持服务器之家。
原文链接:https://www.cnblogs.com/springsnow/p/12017756.html