python网络-多进程(21)
一、什么是进程
进程(process)是计算机中的程序关于某数据集合上的一次运行活动,是系统进行资源分配和调度的基本单位,是操作系统结构的基础。在早期面向进程设计的计算机结构中,进程是程序的基本执行实体;在当代面向线程设计的计算机结构中,进程是线程的容器。程序是指令、数据及其组织形式的描述,进程是程序的实体。---来自百度百科
二、程序和进程的关系
三、fork()
#-*- coding:utf-8 -*- import os pid = os.fork() if pid == 0: print("子进程") else: print("主进程")
运行结果为:
主进程 子进程
getpid()、getppid()
import os pid = os.fork() if pid == 0: print("我是子进程(%d),我的父进程(%d)"%(os.getpid(),os.getppid())) else: print("我是父进程(%d),我的子进程(%d)"%(os.getpid,pid)) print("父子进程都可以执行的代码")
运行结果为:
我是父进程(4488),我的子进程(4491) 父子进程都可以执行的代码 我是子进程(4491),我的父进程(4488) 父子进城都可以执行的代码
说明:
- 程序执⾏到os.fork()时,操作系统会创建⼀个新的进程(⼦进程),然后复制⽗进程的所有信息到⼦进程中
- 然后⽗进程和⼦进程都会从fork()函数中得到⼀个返回值,在⼦进程中这个值⼀定是0,⽽⽗进程中是⼦进程的id号
- 普通的函数调⽤,调⽤⼀次,返回⼀次,但是fork()调⽤⼀次,返回两次,因为操作系统⾃动把当前进程(称为⽗进程)复制了⼀份(称为⼦进程),然后,分别在⽗进程和⼦进程内返回
- ⼀个⽗进程可以fork出很多⼦进程,所以,⽗进程要记下每个⼦进程的id,⽽⼦进程只需要调⽤getppid()就可以拿到⽗进程的id
多个fork()
#-*- coding:utf-8 -*- import os pid1 = os.fork() if pid1 == 0:#子进程1 print("1:我是第一个子进程%d,我的父进程是%d"%(os.getpid(),os.getppid())) else:#父进程 print("2:我是父进程%d"%os.getpid()) pid2 = os.fork() if pid2==0: print("3:我是谁%d,我的父进程是%d"%(os.getpid(),os.getppid())) else: print("4:我是谁%d,我的父进程是%d"%(os.getpid(),os.getppid()))
运行结果为:
2:我是父进程3189 1:我是第一个子进程3190,我的父进程是3189 4:我是谁3190,我的父进程是3189 3:我是谁3191,我的父进程是3189 3:我是谁3192,我的父进程是3190 4:我是谁3189,我的父进程是991
说明:
- pid2开辟的进程将会被子进程1和父进程同时调用
-
当父线程调用pid2
- if pid2 == 0:会在创建一个子进程2,父进程是主进程
- else:及父线程本身,不会再创建进程
-
当子进程1调用pid2
- if pid2 ==0:会创建一个子子进程,父进程是子进程1
- else:即子线程1本身,不会再创建进程
其实上面的代码就相当于:
#-*- coding:utf-8 -*- import os pid1 = os.fork() if pid1 == 0:#子进程1 print("1:我是第一个子进程%d,我的父进程是%d"%(os.getpid(),os.getppid())) else:#父进程 print("2:我是父进程%d"%os.getpid()) pid2 = os.fork() if pid1 == 0:#子进程1 if pid2==0:#子子进程 print("3:我是谁%d,我的父进程是%d"%(os.getpid(),os.getppid())) else:#子进程1 print("4:我是谁%d,我的父进程是%d"%(os.getpid(),os.getppid())) else:#父进程 if pid2==0:#子进程2 print("3:我是谁%d,我的父进程是%d"%(os.getpid(),os.getppid())) else:#父进程 print("4:我是谁%d,我的父进程是%d"%(os.getpid(),os.getppid()))
四、multiprocessing
由于python是跨平台的,自然也应该提供一个跨平台的多进程支持。multiprocessing模块就是跨平台版本的多进程模块。
multiprocessing模块提供了一个process类来代表一个进程对象
#coding=utf-8 from multiprocessing import process import os #子进程要执行的代码 def sub_process(name): print("这是在子进程中,name=%s,pid=%d"%(name,os.getpid())) if __name__ == "__main__": print("父进程:%d"%os.getpid()) p=process(target=sub_process,args=("test",)) print("----子进程将要开启----") p.start()#开启子进程 p.join()#用于等待子进程执行完毕再继续往下执行 print("----子进程已经结束----")
运行结果为:
父进程:8344 ----子进程将要开启---- 这是在子进程中,name=test,pid=9064 ----子进程已经结束----
说明
- 创建子进程时,只需要传入一个执行函数和函数的参数,创建一个process实例,用start()方法启动,这样创建进程比fork()还要简单。
- join()方法可以等待子进程结束后再继续往下运行,通常用于进程间的同步。
process语法结构如下:
process([group [, target [, name [, args [, kwargs]]]]])
1 group:参数未使用,值始终为none 2 target:表示调用对象,即子进程要执行的任务 3 args:表示调用对象的位置参数元组,args=(1,2,'a',) 4 kwargs:表示调用对象的字典,kwargs={'name':'a','age':18} 5 name:为子进程的名称
process类常用方法:
1 start():启动进程,并调用该子进程中的p.run() 2 run():进程启动时运行的方法,正是它去调用target指定的函数,我们自定义类的类中一定要实现该方法 3 terminate():强制终止进程p,不会进行任何清理操作,如果p创建了子进程,该子进程就成了僵尸进程,使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放,进而导致死锁 4 is_alive():如果p仍然运行,返回true 5 join([timeout]):主线程等待p终止(强调:是主线程处于等的状态,而p是处于运行的状态)。timeout是可选的超时时间,需要强调的是,p.join只能join住start开启的进程,而不能join住run开启的进程
process类常用属性:
1 daemon:默认值为false,如果设为true,代表p为后台运行的守护进程,当p的父进程终止时,p也随之终止,并且设定为true后,p不能创建自己的新进程,必须在p.start()之前设置 2 name:进程的名称 3 pid:进程的pid 4 exitcode:进程在运行时为none、如果为–n,表示被信号n结束(了解即可) 5 authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性,这类连接只有在具有相同的身份验证键时才能成功(了解即可)
#coding=utf-8 from multiprocessing import process import time import os #两个子进程将会调用的两个方法 print("1:%d"%os.getpid()) def worker_1(interval): print("worker_1:父进程(%s),当前进程(%s)"%(os.getppid(),os.getpid())) t_start = time.time() time.sleep(interval) #程序将会被挂起interval秒 t_end = time.time() print("worker_1,执行时间为'%0.2f'秒"%(t_end - t_start)) print("2:%d"%os.getpid()) def worker_2(interval): print("worker_2,父进程(%s),当前进程(%s)"%(os.getppid(),os.getpid())) t_start = time.time() time.sleep(interval) t_end = time.time() print("worker_2,执行时间为'%0.2f'秒"%(t_end - t_start)) #输出当前程序的id print("3:%d"%os.getpid()) if __name__=='__main__': print("4:%d"%os.getpid()) p1=process(target=worker_1,args=(2,)) p2=process(target=worker_2,name="se7en_hou",args=(1,)) print("5:%d"%os.getpid()) p1.start() p2.start() #同时父进程仍然往下执行,如果p2进程还在执行,将会返回true print("p2.is_alive=%s"%p2.is_alive()) #输出p1和p2进程的别名和pid print("p1.name=%s"%p1.name) print("p1.pid=%s"%p1.pid) print("p2.name=%s"%p2.name) print("p2.pid=%s"%p2.pid) print("6:%d"%os.getpid()) #join括号中不携带参数,表示父进程在这个位置要等待p1进程执行完成后,再继续执行下面的语句,一般用于进程间的数据同步 p1.join() print("p1.is_alive=%s"%p1.is_alive()) p2.join() print("7:%d"%os.getpid())
运行结果为:
1:10452 2:10452 3:10452 4:10452 5:10452 p2.is_alive=true p1.name=process-1 p1.pid=10688 p2.name=se7en_hou p2.pid=2192 6:10452 1:2192 2:2192 3:2192 7:2192 worker_2,父进程(10452),当前进程(2192) worker_2,执行时间为'1.00'秒 1:10688 2:10688 3:10688 7:10688 worker_1:父进程(10452),当前进程(10688) worker_1,执行时间为'2.00'秒 p1.is_alive=false 7:10452
五、创建process子类创建多进程
创建新的进程还能够使用类的方式,可以自定义一个类,继承process类,每次实例化这个类的时候,就等同于实例化一个进程对象
from multiprocessing import process import time import os #创建一个类,继承process类 class my_process(process): def __init__(self,interval): #因为process类本身也有__init__方法,这个子类相当于重写了这个方法, #但这样就会带来一个问题,我们并没有完全的初始化一个process类,所以就不能使用从这个类继承的一些方法和属性, #最好的方法就是将继承类本身传递给process.__init__方法,完成这些初始化操作 process.__init__(self) self.interval = interval #重写了process类的run()方法 def run(self): print("子进程:%d,开始执行,父进程:%d"%(os.getpid(),os.getppid())) t_start = time.time() time.sleep(self.interval) t_stop = time.time() print("子进程:%d,执行结束,耗时%0.2f秒"%(os.getpid(),t_stop-t_start)) if __name__ == '__main__': t_start = time.time() print("当前进程是%d"%os.getpid()) p1 = my_process(3) p1.start() p1.join() t_stop = time.time() print("当前进程%d执行结束,耗时:%0.2f"%(os.getpid(),t_stop-t_start))
运行结果为:
当前进程是9980 子进程:7084,开始执行,父进程:9980 子进程:7084执行结束,耗时3.00秒 当前进程9980执行结束,耗时:3.23
六、进程池pool
当需要创建的子进程数量不多时,可以直接利用multiprocessing中的process动态成生多个进程,但如果是上百甚至上千个目标,手动的去创建进程的工作量巨大,此时就可以用到multiprocessing模块提供的pool方法。
初始化pool时,可以指定一个最大进程数,当有新的请求提交到pool中时,如果池还没有满,那么就会创建一个新的进程用来执行该请求;但如果池中的进程数已经达到指定的最大值,那么该请求就会等待,直到池中有进程结束,才会创建新的进程来执行
from multiprocessing import pool import os import time import random def worker(msg): t_start = time.time() print("%d进程开始执行%d"%(os.getpid(),msg)) #random.random()随机生成0~1之间的浮点数 time.sleep(random.random()*2) t_stop = time.time() print(msg,"执行完毕,耗时%0.2f"%(t_stop-t_start)) if __name__ == '__main__': po=pool(3) #定义一个进程池,最大进程数3 for i in range(0,10): #pool.apply_async(要调用的目标,(传递给目标的参数元祖,)) #每次循环将会用空闲出来的子进程去调用目标 po.apply_async(worker,(i,)) print("----start----") po.close() #关闭进程池,关闭后po不再接收新的请求 po.join() #等待po中所有子进程执行完成,必须放在close语句之后 print("-----end-----")
运行结果为:
----start---- 4353进程开始执行0 4354进程开始执行1 4355进程开始执行2 2,执行完毕,耗时0.20 4355进程开始执行3 1,执行完毕,耗时1.19 4354进程开始执行4 4,执行完毕,耗时0.37 4354进程开始执行5 0,执行完毕,耗时1.57 4353进程开始执行6 5,执行完毕,耗时0.19 4354进程开始执行7 3,执行完毕,耗时1.63 4355进程开始执行8 6,执行完毕,耗时0.49 4353进程开始执行9 8,执行完毕,耗时0.75 7,执行完毕,耗时0.90 9,执行完毕,耗时0.63 -----end-----
multiprocessing.pool常用函数解析:
- apply_async(func[, args[, kwds]]) :使用非阻塞方式调用func(并行执行,堵塞方式必须等待上一个进程退出才能执行下一个进程),args为传递给func的参数列表,kwds为传递给func的关键字参数列表;
- apply(func[, args[, kwds]]):使用阻塞方式调用func
- close():关闭pool,使其不再接受新的任务;
- terminate():不管任务是否完成,立即终止;
- join():主进程阻塞,等待子进程的退出, 必须在close或terminate之后使用;
apply堵塞式
from multiprocessing import pool import os import time import random def worker(msg): t_start = time.time() print("%d进程开始执行%d"%(os.getpid(),msg)) #random.random()随机生成0~1之间的浮点数 time.sleep(random.random()*2) t_stop = time.time() print(msg,"执行完毕,耗时%0.2f"%(t_stop-t_start)) if __name__ == '__main__': po=pool(3) #定义一个进程池,最大进程数3 for i in range(0,10): #pool.apply_async(要调用的目标,(传递给目标的参数元祖,)) #每次循环将会用空闲出来的子进程去调用目标 po.apply(worker,(i,)) print("----start----") po.close() #关闭进程池,关闭后po不再接收新的请求 po.join() #等待po中所有子进程执行完成,必须放在close语句之后 print("-----end-----")
运行结果为:
4400进程开始执行0 0,执行完毕,耗时1.89 4401进程开始执行1 1,执行完毕,耗时1.91 4402进程开始执行2 2,执行完毕,耗时1.64 4400进程开始执行3 3,执行完毕,耗时1.16 4401进程开始执行4 4,执行完毕,耗时1.85 4402进程开始执行5 5,执行完毕,耗时0.29 4400进程开始执行6 6,执行完毕,耗时0.19 4401进程开始执行7 7,执行完毕,耗时1.19 4402进程开始执行8 8,执行完毕,耗时0.61 4400进程开始执行9 9,执行完毕,耗时1.08 ----start---- -----end-----
说明:通过运行结果可以看出来,阻塞式会等进程池中的进程都执行完毕了才会运行主进程的start和end的打印
七、进程间的通信-queue
1. queue的使用
可以使用multiprocessing模块的queue实现多进程之间的数据传递,queue本身是一个消息列队程序,首先用一个小实例来演示一下queue的工作原理:
#-*- coding:utf-8 -*- from multiprocessing import queue #创建一个queue对象,最多可接受三条put消息 q = queue(3) q.put("消息1") q.put("消息2") print(q.full()) q.put("消息3") print(q.full()) try: q.put("消息4",true,2) except : print("消息队列已满,现有消息数量:%s"%q.qsize()) try: q.put_nowait("消息5") except : print("消息队列已满,现有消息数量:%s"%q.qsize()) #推荐方式,先判断消息队列是否已满,在写入 if not q.full(): q.put_nowait("消息6") #读取消息时,先判断消息队列是否为空,在读取 if not q.empty(): for i in range(q.qsize()): print(q.get_nowait())
运行结果为:
false true 消息队列已满,现有消息数量:3 消息队列已满,现有消息数量:3 消息1 消息2 消息3
说明
初始化queue()对象时(例如:q=queue()),若括号中没有指定最大可接收的消息数量,或数量为负值,那么就代表可接受的消息数量没有上限(直到内存的尽头);
- queue.qsize():返回当前队列包含的消息数量;
- queue.empty():如果队列为空,返回true,反之false ;
- queue.full():如果队列满了,返回true,反之false;
- queue.get([block[, timeout]]):获取队列中的一条消息,然后将其从列队中移除,block默认值为true;
1)如果block使用默认值,且没有设置timeout(单位秒),消息列队如果为空,此时程序将被阻塞(停在读取状态),直到从消息列队读到消息为止,如果设置了timeout,则会等待timeout秒,若还没读取到任何消息,则抛出"queue.empty"异常;
2)如果block值为false,消息列队如果为空,则会立刻抛出"queue.empty"异常;
-
queue.get_nowait():相当queue.get(false);
-
queue.put(item,[block[, timeout]]):将item消息写入队列,block默认值为true;
1)如果block使用默认值,且没有设置timeout(单位秒),消息列队如果已经没有空间可写入,此时程序将被阻塞(停在写入状态),直到从消息列队腾出空间为止,如果设置了timeout,则会等待timeout秒,若还没空间,则抛出"queue.full"异常;
2)如果block值为false,消息列队如果没有空间可写入,则会立刻抛出"queue.full"异常;
- queue.put_nowait(item):相当queue.put(item, false);
2. queue实例
我们以queue为例,在父进程中创建两个子进程,一个往queue里写数据,一个从queue里读数据:
from multiprocessing import process from multiprocessing import queue import os import time import random #写数据进程执行的代码 def write(q): for value in ["a","b","c"]: print("put %s to queue "%value) q.put(value) time.sleep(random.random()) #读取数据进程的代码 def read(q): while true: if not q.empty(): value = q.get(true) print("get %s from queue "%value) time.sleep(random.random()) else: break if __name__ == '__main__': #父进程创建queue,并传递给各个子进程 q = queue() pw = process(target = write,args=(q,)) pr = process(target = read,args=(q,)) #启动子进程pw,写入 pw.start() #等待pw结束 pw.join() #启动子进程pr,读取 pr.start() pr.join() print("所有数据都写入并且读完")
运行结果为:
put a to queue put b to queue put c to queue get a from queue get b from queue get c from queue 所有数据都写入并且读完
3. 进程池中的queue
如果要使用pool创建进程,就需要使用multiprocessing.manager()中的queue(),而不是multiprocessing.queue(),否则会得到一条如下的错误信息:
runtimeerror: queue objects should only be shared between processes through inheritance.
#coding=utf-8 from multiprocessing import manager from multiprocessing import pool import os import time import random def reader(q): print("reader启动(%d),父进程为(%d)"%(os.getpid(),os.getppid())) for i in range(q.qsize()): print("reader从queue获取到的消息时:%s"%q.get(true)) def writer(q): print("writer启动(%d),父进程为(%d)"%(os.getpid(),os.getppid())) for i in "se7en_hou": q.put(i) if __name__ == '__main__': print("-------(%d) start-------"%os.getpid()) #使用manager中的queue来初始化 q = manager().queue() po = pool() #使用阻塞模式创建进程,这样就不需要在reader中使用死循环了,可以让writer完全执行完成后,再用reader去读取 po.apply(writer,(q,)) po.apply(reader,(q,)) po.close() po.join() print("-------(%d) end-------"%os.getpid())
运行结果为:
-------(880) start------- writer启动(7744),父进程为(880) reader启动(7936),父进程为(880) reader从queue获取到的消息时:s reader从queue获取到的消息时:e reader从queue获取到的消息时:7 reader从queue获取到的消息时:e reader从queue获取到的消息时:n reader从queue获取到的消息时:_ reader从queue获取到的消息时:h reader从queue获取到的消息时:o reader从queue获取到的消息时:u -------(880) end-------