Python全栈开发-Day10-进程/协程/异步IO/IO多路复用
Posted GavinSimons
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Python全栈开发-Day10-进程/协程/异步IO/IO多路复用相关的知识,希望对你有一定的参考价值。
本节内容
- 多进程multiprocessing
- 进程间的通讯
- 协程
- 论事件驱动与异步IO
- Select\\Poll\\Epoll——IO多路复用
1、多进程multiprocessing
Python的线程用的是操作系统的原生线程,同样python的进程用的是操作系统的原生进程。
多进程之间没有锁的概念,多进程之间数据不能互相访问,所以不存在互斥锁。GIL问题又是仅仅出现在多线程中。
所以如果我们启动8个进程,每个进程有一个主线程,即8个线程,分别运行在8个CPU上,就可以充分利用多核的优势了。
在多进程充分利用多核的优势下,唯一的坏处是这8个进程之间数据无法共享。传递数据需要找媒介。
8进程表示,同一时间最多只能干8件事情。
所以多进程可以解决多核的问题
同时每个进程里又可以写多个线程,启动并执行。
每一个进程都是由它的父进程启动的。
os.getppid() #获得父进程的id
os.getpid() #获得自己进程的id
1
2
3
4
5
6
7
8
9
10
|
from multiprocessing import Process import time def f(name): time.sleep( 2 ) print ( \'hello\' , name) if __name__ = = \'__main__\' : p = Process(target = f, args = ( \'bob\' ,)) p.start() p.join() |
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
from multiprocessing import Process import os def info(title): print (title) print ( \'module name:\' , __name__) print ( \'parent process:\' , os.getppid()) print ( \'process id:\' , os.getpid()) print ( "\\n\\n" ) def f(name): info( \'\\033[31;1mfunction f\\033[0m\' ) print ( \'hello\' , name) if __name__ = = \'__main__\' : info( \'\\033[32;1mmain process line\\033[0m\' ) p = Process(target = f, args = ( \'bob\' ,)) p.start() p.join() |
2、进程间的通讯
不同进程间内存是不共享的,要想实现两个进程间的数据交换,可以用以下方法:
Queues(进程队列)——数据传递
使用方法跟threading里的queue差不多,
原理上是两个进程各有一个queue队列,通过pickle序列化的方式,实现两个队列之间的交互,从而看起来像两个进程之间的交互。
所以实际上进程队列是2个Queues,而不是一个共享queue队列。
1
2
3
4
5
6
7
8
9
10
11
12
|
from multiprocessing import Process, Queue def f(q): q.put([ 42 , None , \'hello\' ]) if __name__ = = \'__main__\' : q = Queue() p = Process(target = f, args = (q,)) p.start() print (q.get()) # prints "[42, None, \'hello\']" p.join() |
Pipes(管道)——数据传递
pipe管道就相当于拿了一根电话线,两头分别连接了两个进程。原理是利用socket网络协议进行周转。
pipe实例一生成会产生两个返回对象,一个是管道的一头,另一个是管道的另一头。
1
2
3
4
5
6
7
8
9
10
11
12
|
from multiprocessing import Process, Pipe def f(conn): conn.send([ 42 , None , \'hello\' ]) conn.close() if __name__ = = \'__main__\' : parent_conn, child_conn = Pipe() p = Process(target = f, args = (child_conn,)) p.start() print (parent_conn.recv()) # prints "[42, None, \'hello\']" p.join() |
Managers已经自动加锁,所以不需要手动加锁。
A manager object returned by Manager()
controls a server process which holds Python objects and allows other processes to manipulate them using proxies.
A manager returned by Manager()
will support types list
, dict
, Namespace
, Lock
, RLock
, Semaphore
, BoundedSemaphore
, Condition
, Event
, Barrier
, Queue
, Value
and Array
. For example,
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
|
from multiprocessing import Process, Manager def f(d, l): d[ 1 ] = \'1\' d[ \'2\' ] = 2 d[ 0.25 ] = None l.append( 1 ) print (l) if __name__ = = \'__main__\' : with Manager() as manager: d = manager. dict () l = manager. list ( range ( 5 )) p_list = [] for i in range ( 10 ): p = Process(target = f, args = (d, l)) p.start() p_list.append(p) for res in p_list: res.join() print (d) print (l) |
进程锁
虽然本身由于进程间不能共享数据,所以不需要锁。但是当进程往屏幕上打印数据时,对各个进程而言屏幕是共享的。所以这个锁的目的是锁住一个时间只能一个进程去打印数据。保证打印数据的完整性,前面正在打印的数据,不被后面要打印的数据打断。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
from multiprocessing import Process, Lock def f(l, i): l.acquire() try : print ( \'hello world\' , i) finally : l.release() if __name__ = = \'__main__\' : lock = Lock() for num in range ( 10 ): Process(target = f, args = (lock, num)).start() |
进程池
进程池内部维护一个进程序列,当使用时,则去进程池中获取一个进程,如果进程池序列中没有可供使用的进程,那么程序就会等待,直到进程池中有可用进程为止。
进程池中有两个方法:
- apply #进程池串行
- apply_async #进程池并行
对于进程池中进程的启动是要使用pool.apply()或pool.apply_async()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
from multiprocessing import Process,Pool,freeze_support import time def Foo(i): time.sleep( 2 ) return i + 100 def Bar(arg): print ( \'-->exec done:\' ,arg) if __name__ == \'__main__\':
freeze_support() #在windows上必须添加这行和上行代码
pool = Pool( 5 ) #允许进程池里同时放入5个进程 for i in range ( 10 ): pool.apply_async(func = Foo, args = (i,),callback = Bar) #
pool.apply_async(func = Foo, args = (i,) ) #pool.apply(func=Foo, args=(i,)) print ( \'end\' ) pool.close() pool.join() #进程池中进程执行完毕后再关闭,如果注释,那么程序直接关闭。 |
这里不知何种原因,必须是先pool.close(),然后再是pool.join()
如果最后不加pool.join(),主程序不会等待进程池执行完毕,会直接关闭。
if __name__ == \'__main__\': #这句话是为了区分,主动启动脚本还是把它当成一个模块,从别的地方去调用。
如果主动启动该脚本,则该段代码下面部分代码会被执行。如果从别的地方调用,则下面代码不会被执行。
主动执行该py文件时,把这句话当成主程序的入口。
上面代码中含有pool.apply_async(func
=
Foo, args
=
(i,),callback
=
Bar),callback叫做回调,意思是执行完这个语句后再回调Bar函数。这里需要注意,callback方法的调用进程是主进程,而不是子进程。
回调函数的意义,如果开了100个子进程,在每个子进程结束时往数据库中插入数据,需要建立100个连接,但如果先把100个子进程的结果保存到变量里,让父进程连接数据库一次,然后一口气全部插入,就大大提高了程序的运行速度。此时的回调函数,应运而生。
3、协程
协程,又称微线程。英文名Coroutine。一句话说明什么是线程:协程是一种用户态的轻量级线程。
CPU只认识线程,并不知道协程的存在。协程是跑在线程中的。
协程拥有自己的寄存器上下文和栈。协程调度切换时,将寄存器上下文和栈保存到其他地方,在切回来的时候,恢复先前保存的寄存器上下文和栈。因此:
协程能保留上一次调用时的状态(即所有局部状态的一个特定组合),每次过程重入时,就相当于进入上一次调用的状态,换种说法:进入上一次离开时所处逻辑流的位置。
换句话说:在单线程下,实现并发的效果,就是协程。例如,之前使用yield做的生产者消费者模型。
协程的好处:
- 无需线程上下文切换的开销
- 无需原子操作锁定及同步的开销
- "原子操作(atomic operation)是不需要synchronized",所谓原子操作是指不会被线程调度机制打断的操作;这种操作一旦开始,就一直运行到结束,中间不会有任何 context switch (切换到另一个线程)。原子操作可以是一个步骤,也可以是多个操作步骤,但是其顺序是不可以被打乱,或者切割掉只执行部分。视作整体是原子性的核心。
- 方便切换控制流,简化编程模型
- 高并发+高扩展性+低成本:一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理。
缺点:
- 无法利用多核资源:协程的本质是个单线程,它不能同时将单个CPU 的多个核用上,协程需要和进程配合才能运行在多CPU上.当然我们日常所编写的绝大部分应用都没有这个必要,除非是cpu密集型应用。
- 进行阻塞(Blocking)操作(如IO时)会阻塞掉整个程序
使用yield实现协程操作的例子
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
|
import time import queue def consumer(name): print ( "--->starting eating baozi..." ) while True : new_baozi = yield #yield这里本身可以返回数据,也可接受数据 print ( "[%s] is eating baozi %s" % (name,new_baozi)) #time.sleep(1) def producer(): r = con.__next__() r = con2.__next__() n = 0 while n < 5 : n + = 1 con.send(n) #激活yield的阻塞状态,同时传入一个数据 con2.send(n) print ( "\\033[32;1m[producer]\\033[0m is making baozi %s" % n ) if __name__ = = \'__main__\' : con = consumer( "c1" ) con2 = consumer( "c2" ) p = producer() |
协程之所以能处理大并发,就是把I/O操作给挤掉了,即一旦遇到I/O操作就切换。使得整个程序变成了只有CPU运算,大大提高了效率。
只要I/O操作一完成,CPU就可以切换回去了。这样就把I/O操作完全挤出去了。程序会自动检测I/O是否完成,不需要我们关心。
协程的标准定义,即符合什么条件就能称之为协程:
- 必须在只有一个单线程里实现并发
- 修改共享数据不需加锁
- 用户程序里自己保存多个控制流的上下文栈
- 一个协程遇到IO操作自动切换到其它协程
Greenlet
yield是自己写的协程,greenlet是一个封装好的协程。
greenlet是一个用C实现的协程模块,相比与python自带的yield,它可以使你在任意函数之间随意切换,而不需把这个函数先声明为generator
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
# -*- coding:utf-8 -*- from greenlet import greenlet def test1(): print ( 12 ) gr2.switch() #gr2.switch()的意思是切换到gr2 print ( 34 ) gr2.switch() #同上 def test2(): print ( 56 ) gr1.switch() print ( 78 ) gr1 = greenlet(test1) #启动一个协程 gr2 = greenlet(test2) gr1.switch() #gr1.switch()的意思是切换到gr1 |
感觉确实用着比generator还简单了呢,但好像还没有解决一个问题,就是遇到IO操作,自动切换,对不对?
greenlet只能手动切换,就相当于汽车的手动档。
Gevent
Gevent 是一个第三方库,可以轻松通过gevent实现并发同步或异步编程,在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。 Greenlet全部运行在主程序操作系统进程的内部,但它们被协作式地调度。
【注意】上文说greenlet手动切换,相当于手动档汽车,而gevent是自动切换,相当于自动档汽车。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
import gevent def func1(): print ( \'\\033[31;1m李闯在跟海涛搞...\\033[0m\' ) gevent.sleep( 2 ) print ( \'\\033[31;1m李闯又回去跟继续跟海涛搞...\\033[0m\' ) def func2(): print ( \'\\033[32;1m李闯切换到了跟海龙搞...\\033[0m\' ) gevent.sleep( 1 ) print ( \'\\033[32;1m李闯搞完了海涛,回来继续跟海龙搞...\\033[0m\' ) gevent.joinall([ gevent.spawn(func1), #生成 gevent.spawn(func2), #gevent.spawn(func3), ]) |
通过gevent实现单线程下的多socket并发
server side
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
|
import sys import socket import time import gevent from gevent import socket,monkey monkey.patch_all() def server(port): s = socket.socket() s.bind(( \'0.0.0.0\' , port)) s.listen( 500 ) while True : cli, addr = s.accept() gevent.spawn(handle_request, cli) def handle_request(conn): try : while True : data = conn.recv( 1024 ) print ( "recv:" , data) conn.send(data) if not data: conn.shutdown(socket.SHUT_WR) except Exception as ex: print (ex) &n |