标签:
一. 进程
multiprocessing包是Python中的多进程管理包,是一个跨平台版本的多进程模块。与threading.Thread类似,它可以利用multiprocessing.Process对象来创建一个进程。该进程可以运行在Python程序内部编写的函数。该Process对象与Thread对象的用法类似。
创建一个Process实例,可用start()方法启动。
join()方法可以等待子进程结束后再继续往下运行,通常用于进程间的同步。
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() |
写个程序,对比下主进程和子进程的ID:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
from multiprocessing import Process import os def info(title): print (title) print ( ‘进程名称:‘ , __name__) print ( ‘父进程ID:‘ , os.getppid()) print ( ‘子进程ID:‘ , os.getpid()) print ( "\n\n" ) def f(name): info( ‘\033[31;1mcalled from child process function 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() |
2. 进程间通信
不同进程间内存是不共享的,要想实现两个进程间的数据交换,可以使用Queue、Pipe、Manager,其中:
1)Queue \ Pipe 只是实现进程间数据的传递;
2)Manager 实现了进程间数据的共享,即多个进程可以修改同一份数据;
Queue允许多个进程放入,多个进程从队列取出对象,先进先出。(使用方法跟threading里的queue差不多)
1
2
3
4
5
6
7
8
9
10
11
12
|
from multiprocessing import Process,Queue def f(qq): qq.put([ 42 , None , "hello" ]) qq.put([ 43 , None , "HI" ]) if __name__ = = ‘__main__‘ : q = Queue() p = Process(target = f,args = (q,)) p.start() print (q.get()) print (q.get()) p.join() |
Pipe也是先进先出
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
from multiprocessing import Process, Pipe def f(conn): conn.send([ 42 , None , ‘儿子发送的消息‘ ]) conn.send([ 42 , None , ‘儿子又发消息啦‘ ]) print ( "接收父亲的消息:" ,conn.recv()) 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‘]" print (parent_conn.recv()) # prints "[42, None, ‘hello‘]" parent_conn.send( "回家吃饭!" ) # prints "[42, None, ‘hello‘]" p.join() |
Manager对象类似于服务器与客户之间的通信 (server-client),与我们在Internet上的活动很类似。我们用一个进程作为服务器,建立Manager来真正存放资源。其它的进程可以通过参数传递或者根据地址来访问Manager,建立连接后,操作服务器上的资源。在防火墙允许的情况下,我们完全可以将Manager运用于多计算机,从而模仿了一个真实的网络情境。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
from multiprocessing import Process,Manager import os def f(d,l): d[os.getpid()] = os.getpid() l.append(os.getpid()) 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() |
3. 进程池
进程池 (Process Pool)可以创建多个进程。这些进程就像是随时待命的士兵,准备执行任务(程序)。一个进程池中可以容纳多个待命的士兵。
进程池有两种方法:
1)串行:apply
2)并行:apply_async
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
from multiprocessing import Process,Pool import time import os def Foo(i): time.sleep( 2 ) print ( "in process" ,os.getpid()) return i + 100 def Bar(arg): ‘‘‘回调函数‘‘‘ print ( "-->>exec done:" ,arg,os.getpid()) if __name__ = = "__main__" : pool = Pool(processes = 3 ) #允许进程池同时放入3个进程 print ( "主进程" ,os.getpid()) for i in range ( 10 ): pool.apply_async(func = Foo,args = (i,),callback = Bar) print ( ‘end‘ ) pool.close() pool.join() #进程池中进程执行完毕后在关闭;如果注释则程序直接关闭 |
使用回调函数的目的是:在父进程中执行可以提高效率;(比如连接数据库,写回调函数的话,父进程连接一次数据库即可;如果使用子进程,则需要连接多次)
4. 其他(lock)
lock:屏幕上打印的锁,防止打印显示混乱
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
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() |
二. 协程
协程,又称微线程,纤程。英文名Coroutine。
协程拥有自己的寄存器上下文和栈。协程调度切换时,将寄存器上下文和栈保存到其他地方,在切回来的时候,恢复先前保存的寄存器上下文和栈。因此:协程能保留上一次调用时的状态(即所有局部状态的一个特定组合),每次过程重入时,就相当于进入上一次调用的状态,换种说法:进入上一次离开时所处逻辑流的位置。
好处:
缺点:
传统的生产者-消费者模型是一个线程写消息,一个线程取消息,通过锁机制控制队列和等待,但一不小心就可能死锁。
如果改用协程,生产者生产消息后,直接通过yield
跳转到消费者开始执行,待消费者执行完毕后,切换回生产者继续生产,效率极高。
代码示例:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
def consumer(): r = ‘‘ while True : n = yield r if not n: return print ( ‘[消费者] Consuming %s...‘ % n) r = ‘200 OK‘ def produce(c): c.send( None ) n = 0 while n < 5 : n = n + 1 print ( ‘[生产者] Producing %s...‘ % n) r = c.send(n) print ( ‘[生产者] 消费者返回状态码: %s‘ % r) c.close() c = consumer() produce(c) |
输出结果:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
[生产者] Producing 1. .. [消费者] Consuming 1. .. [生产者] 消费者返回状态码: 200 OK [生产者] Producing 2. .. [消费者] Consuming 2. .. [生产者] 消费者返回状态码: 200 OK [生产者] Producing 3. .. [消费者] Consuming 3. .. [生产者] 消费者返回状态码: 200 OK [生产者] Producing 4. .. [消费者] Consuming 4. .. [生产者] 消费者返回状态码: 200 OK [生产者] Producing 5. .. [消费者] Consuming 5. .. [生产者] 消费者返回状态码: 200 OK |
注意到consumer
函数是一个generator
,把一个consumer
传入produce
后:
c.send(None)
启动生成器;c.send(n)
切换到consumer
执行;consumer
通过yield
拿到消息,处理,又通过yield
把结果传回;produce
拿到consumer
处理的结果,继续生产下一条消息;produce
决定不生产了,通过c.close()
关闭consumer
,整个过程结束。整个流程无锁,由一个线程执行,生产者
和消费者
协作完成任务,所以称为“协程”,而非线程的抢占式多任务。(原理:遇到I/O操作就切换,只剩下CPU操作(CPU操作非常快))
一句话总结协程的特点:子程序就是协程的一种特例。
python中支持协程的有以下两个模块:greenlet和greent
greenlet封装好的协程,利用.swith对协程操作进行手动切换
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
from greenlet import greenlet def test1(): print ( 12 ) gr3.switch() print ( 34 ) gr2.switch() print ( 78 ) def test2(): print ( 56 ) gr1.switch() def test3(): print ( 90 ) gr1.switch() gr1 = greenlet(test1) #启动协程 gr2 = greenlet(test2) gr3 = greenlet(test3) gr1.switch() |
3. Greent
Gevent 是一个第三方库,可以轻松通过gevent实现并发同步或异步编程,在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。 Greenlet全部运行在主程序操作系统进程的内部,但它们被协作式地调度。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
import gevent def foo(): print ( "运行foo" ) gevent.sleep( 2 ) print ( "再次回到foo" ) def bar(): print ( "这里是bar" ) gevent.sleep( 1 ) print ( "又回到了bar" ) def func3(): print ( "运行func3" ) gevent.sleep( 0 ) print ( "再次运行func3" ) gevent.joinall([ gevent.spawn(foo), gevent.spawn(bar), gevent.spawn(func3) ]) |
同步与异步的性能区别:
1)同步:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
from gevent import monkey; # monkey.patch_all() import gevent from urllib.request import urlopen import time def f(url): print ( ‘GET: %s‘ % url) resp = urlopen(url) data = resp.read() print ( ‘%d bytes received from %s.‘ % ( len (data), url)) urls = [ ‘https://www.python.org/‘ , ‘https://www.yahoo.com/‘ , ‘https://github.com/‘ ] time_start = time.time() for url in urls: f(url) print ( "同步cost" ,time.time() - time_start) |
2)异步:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
from gevent import monkey; # monkey.patch_all() import gevent from urllib.request import urlopen import time def f(url): print ( ‘GET: %s‘ % url) resp = urlopen(url) data = resp.read() print ( ‘%d bytes received from %s.‘ % ( len (data), url)) urls = [ ‘https://www.python.org/‘ , ‘https://www.yahoo.com/‘ , ‘https://github.com/‘ ] async_time_start = time.time() gevent.joinall([ gevent.spawn(f, ‘https://www.python.org/‘ ), gevent.spawn(f, ‘https://www.yahoo.com/‘ ), gevent.spawn(f, ‘https://github.com/‘ ), ]) print ( "异步cost" ,time.time() - async_time_start ) |
结论:同步开销时间为4秒,异步开销为2.5秒,大大节省了开销,这就是协程的魅力;monkey.patch_all()使gevent能识别到urllib中的I/O操作
使用gevent实现单线程下的多socket并发:
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) finally: conn.close() if __name__ == ‘__main__‘: server(8001)
import socket HOST = ‘localhost‘ # The remote host PORT = 8001 # The same port as used by the server s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.connect((HOST, PORT)) while True: msg = bytes(input(">>:"),encoding="utf8") s.sendall(msg) data = s.recv(1024) #print(data) print(‘Received‘, repr(data)) s.close()
标签:
原文地址:http://www.cnblogs.com/liuyuchen123456/p/5959827.html