multiprocessing摸快介绍
python中的多线程无法利用多核优势,如果想要充分地使用多核CPU的资源(os.cpu_count()查看),在python中大部分情况需要使用多进程。Python提供了multiprocessing。
multiprocessing模块用来开启子进程,并在子进程中执行我们定制的任务(比如函数),该模块与多线程模块threading的编程接口类似。
multiprocessing模块的功能众多:支持子进程、通信和共享数据、执行不同形式的同步,提供了Process、Queue、Pipe、Lock等组件。
需要再次强调的一点是:与线程不同,进程没有任何共享状态,进程修改的数据,改动仅限于该进程内。
Process类的介绍
创建进程的类:
Process([group [, target [, name [, args [, kwargs]]]]]),由该类实例化得到的对象,表示一个子进程中的任务(尚未启动) 强调: 1. 需要使用关键字的方式来指定参数 2. args指定的为传给target函数的位置参数,是一个元组形式,必须有逗号
参数介绍:
group 参数未使用,值始终为None target 表示调用对象,即子进程要执行的任务 args 表示调用对象的位置参数元组,args=(1,2,'egon',) kwargs 表示调用对象的字典,kwargs={'name':'egon','age':18} name 为子进程的名称
方法介绍:
1 p.start() 启动进程,并调用该子进程中的p.run() 2 p.run() 进程启动时运行的方法,正是它去调用target指定的函数,我们自定义类的类中一定要实现该方法 3 p.terminate() 强制终止进程p,不会进行任何清理操作,如果p创建了子进程,该子进程就成了僵尸进程,使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放,进而导致死锁 4 p.is_alive() 如果p仍然运行,返回True 5 p.join([timeout]) 主线程等待p终止(强调:是主线程处于等的状态,而p是处于运行的状态)。timeout是可选的超时时间,需要强调的是,p.join只能join住start开启的进程,而不能join住run开启的进程
属性介绍:
1 p.daemon 默认值为False,如果设为True,代表p为后台运行的守护进程,当p的父进程终止时,p也随之终止,并且设定为True后,p不能创建自己的新进程,必须在p.start()之前设置 2 p.name 进程的名称 3 p.pid 进程的pid 4 p.exitcode 进程在运行时为None、如果为–N,表示被信号N结束(了解即可) 5 p.authkey 进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性,这类连接只有在具有相同的身份验证键时才能成功(了解即可)
Process类的使用
注意:在Windows中,Process()必须放在 if __name__ == "__main__": 下执行
一个python文件通常有两种使用方法:一个是作为脚本执行,一个是import到其他文件中被调用执行。 因此 if __name__ == 'main': 的作用就是控制这两种情况执行代码的过程,在 if __name__ == 'main': 下的代码只有在第一种情况下(即文件作为脚本直接执行)才会被执行,而 import 到其他脚本中是不会被执行的。 每个python模块/文件都包含内置变量__name__。当该模块作为脚本执行时,__name__==文件名.py;当该模块import到其他模块中时,__name__==文件名。而__main__始终等于文件名.py。 所以,当文件当做脚本执行时,if为真,才会执行
创建并开启子进程的两种方法:
from multiprocessing import Process def func(name): print("my name is %s"%name) p = Process(target=func,args=("张三",)) # 必须加逗号,因为要传 元组 p.start() from multiprocessing import Process class Cls(Process): def __init__(self,name): super().__init__() self.name = name def run(self): print("my name is %s"%name) p = Cls("张三") p.start()
进程之间的内存空间是隔离的
from multiprocessing import Process n=100 def work(): global n n=0 print('子进程内: ',n) if __name__ == '__main__': p=Process(target=work) p.start() print('主进程内: ',n) # 主进程与子进程n的值不同,说明 进程间的内存空间是隔离的
Process对象的join方法
from multiprocessing import Process def func(name): print("my name is %s"%name) p1 = Process(target=func,args=("张三",)) p2 = Process(target=func,args=("李四",)) p1.start() p2.start() p1.join() p2.join() print("主进程") 程序中添加了join,变成串行了吗? 不是,join只是让主进程等待子进程结束,卡住的是主进程而不是子进程。主进程等待的时候,各个子进程仍在执行。执行花费的总时间是 子进程中耗时最长的那个。
Process对象的其他方法或属性(了解),name,pid,is_alive,terminate
from multiprocessing import Process import time def func(name): print("my name is %s"%name) time.sleep(3) print("my name is %s"%name) p = Process(target=func,args=("张三",),name="进程名字") p.start()
print(p.name) # 进程名字 print(p.pid) # 进程pid
print(p.is_alive()) # 判断进程是否在运行,True p.terminate() # 关闭进程,但不会立即关闭 time.sleep(1) print(p.is_alive()) # False
思考
from multiprocessing import Process def func(name): print("my name is %s"%name) p = Process(target=func,args=("张三",)) p.start() p.join() # 等待进程p结束后,join函数内部会发送系统调用wait,去告诉操作系统回收进程的pid print(p.pid) # 能否打印进程的pid 可以 p.join()是向操作系统发送请求,告知操作系统p的id号不需要再占用了,回收掉 此时,在父进程中还可以看到p.pid,但此时p.pid是一个无意义的id号,因为操作系统已经将编号回收
守护进程
主进程创建守护进程。
守护进程会在主进程执行结束后终止
守护进程内无法再开启子进程,否则会抛出异常:AssertionError: daemonic processes are not allowed to have children
注意:进程间是相互独立的,主进程代码运行结束,守护进程也随即终止
from multiprocessing import Process def func(name): print("my name is %s"%name) p = Process(target=func,args=("张三",)) p.daemon = True p.start() 守护进程,一定要在p.start()之前设置,禁止p再创建子进程
进程同步(锁)
进程间数据不共享,但是共享一套文件系统,所以访问同一个文件是没有问题的。但当修改文件时,就会出现错误,此时,可以加锁处理
#由并发变成了串行,牺牲了运行效率,但避免了竞争 from multiprocessing import Process,Lock import os,time def work(lock): lock.acquire() print('%s is running' %os.getpid()) time.sleep(2) print('%s is done' %os.getpid()) lock.release() if __name__ == '__main__': lock=Lock() for i in range(3): p=Process(target=work,args=(lock,)) p.start()
总结:
加锁可以保证多个进程修改同一数据时,同一时间只能有一个任务可以修改,即串行的修改,但是 牺牲了速度却保证了数据安全
因此,我们需要的解决问题方法,同时兼顾高效率(多个进程共享一块内存的数据),并且帮我们处理好锁的问题。
这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制:队列和管道
1.队列和管道,都是将数据存放在内存中
2.队列是基于 管道+锁 实现的,可以让我们从负责的锁问题中解脱出来
我们应该尽量避免使用共享数据,尽可能使用消息传递和队列,避免处理复杂的同步和锁问题,而且在进程数目增多时,往往可以获得更好的可扩展性。
队列(推荐使用) Queue
进程彼此之间互相隔离,要实现进程间通信(IPC),multiprocessing模块支持两种形式:队列和管道,这两种方式都是使用消息传递的。
创建队列的类(底层就是以管道和锁定的方式实现):
Queue([maxsize]):创建共享的进程队列,Queue是多进程安全的队列,可以使用Queue实现多进程之间的数据传递,maxsize队列中允许最大项目,不设置则没有大小限制
方法介绍:
1 q.put方法用以插入数据到队列中,put方法还有两个可选参数:block和timeout。
如果block为True(默认值),并且timeout为正值,该方法会阻塞timeout所指定的时间,直到该队列有剩余的空间。如果超时,会抛出Queue.Full异常。如果block为False,但该Queue已满,会立即抛出Queue.Full异常。 2 q.get方法可以从队列中读取并删除一个元素。同样,get方法也有两个可选参数:blocked和timeout。
如果blocked为True(默认),并且timeout为正值,那么在等待时间内没有取到任何元素,会抛出Queue.Empty异常。如果blocked为False,有两种情况存在,如果Queue有一个值可用,则立即返回,否则,如果队列为空,则立即抛出Queue.Empty异常。 3 q.get_nowait() == q.get(False) 4 q.put_nowait() == q.put(False) 5 q.empty() 判断q是否为空,该方法不可靠,因为返回结果的过程中,可能会对q进行操作 6 q.full() 判断q是否已满,该方法不可靠,因为返回结果的过程中,可能会对q进行操作 7 q.cancel_join_thread() 不会在进程退出时自动连接后台线程,可以防止join_thread()方法阻塞。 8 q.close() 关闭队列。调用此方法,后台线程将继续写入那些已经入列但尚未写入的数据,但将在此方法执行完成时马上关闭。
如果q被垃圾收集,将调用此方法。关闭队列不会在队列使用者中产生任务类型的数据结束信号或异常。比如:如果某个使用者正阻塞在get()操作上,关闭生产者中的队列不会导致get()方法返回错误。 9 q.join_thread(): 连接队列的后台线程。此方法用于调用q.close()方法后,等待所有队列项被消耗。默认情况下,此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread方法可以禁止这种行为。
from multiprocessing import Process,Queue q = Queue(3) # q只能存放三个元素 q.put(1) q.put(2) q.put(3) print(q.full()) # q满了 print(q.get()) # 先进先出,按顺序取 print(q.get()) print(q.get()) print(q.empty()) # q空了
管道 Pipe
进程间通信(IPC)的另一种方式,推荐使用 队列Queue
创建管道的类:
Pipe([dumplex]),在进程之间创建一条管道,并返回元组(conn1,conn2),其中conn1,conn2表示管道两段的连接对象,但 必须在产生Process对象之前产生管道
参数介绍:
duplex:默认管道是全双工的,如果将duplex设成False,conn1只能用于接收,conn2只能用于发送。
主要方法:
conn1.recv():接收conn2.send(obj)发送的对象,如果没有消息可接收,recv方法会一直阻塞,如果连接的另外一端已经关闭,那么recv方法会抛出EOFError。
conn1.send(obj): 通过链接发送对象,obj是与序列化兼容的任意对象
其他方法:
conn1.close(): 关闭连接,如果conn1被垃圾回收,将自动调用此方法
conn1.fileno():返回连接使用的整数文件描述
conn1.poll([timeout]):如果连接上的数据可用,返回True,不设置timeout则立即返回结果,设置为None,将一直等待数据到达
conn1.recv_bytes([maxlength]):接收s.send_bytes()方法发送的一条完整的字节消息。maxlength指定要接收的最大字节数。如果进入的消息超过了这个最大值,将引发IOError异常,并且在连接上无法进行进一步读取,如果连接的另外一端已经关闭,再也不存在任何数据,将引发EOFError异常。
conn.send_bytes(buffer[,offset[,size]]):通过连接发送字节数据缓冲区,buffer是支持缓冲区接口的任意对象,offset是缓冲区中的字节偏移量,而size是要发送字节数。结果数据以单条消息的形式发出,然后调用c.recv_bytes()函数进行接收。
conn1.recv_bytes_into(buffer[,offset]):接收一条完整的字节消息,并把它保存在buffer对象中,该对象支持可写入的缓冲区接口(即bytearray对象或类似的对象)。offset指定缓冲区中放置消息处的字节位移,返回值是收到的字节数。如果消息长度大于可用的缓冲区空间,将引发BufferTooShort异常。
from multiprocessing import Process,Pipe import time,os def consumer(p,name): left,right=p left.close() while True: try: baozi=right.recv() print('%s 收到包子:%s' %(name,baozi)) except EOFError: right.close() break def producer(seq,p): left,right=p right.close() for i in seq: left.send(i) # time.sleep(1) else: left.close() if __name__ == '__main__': left,right=Pipe() c1=Process(target=consumer,args=((left,right),'c1')) c1.start() seq=(i for i in range(10)) producer(seq,(left,right)) right.close() left.close() c1.join() print('主进程')
注意:生产者和消费者都没有使用管道的某个端点,就应该将其关闭。如在生产者中关闭管道的右端,在消费者中关闭管道的左端。如果忘记执行这些步骤,程序可能再消费者中的recv()操作上挂起。管道是由操作系统进行引用计数的,必须在所有进程中关闭管道后才能生产EOFError异常。因此在生产者中关闭管道不会有任何效果,付费消费者中也关闭了相同的管道端点。
from multiprocessing import Process,Pipe import time,os def adder(p,name): server,client=p client.close() while True: try: x,y=server.recv() except EOFError: server.close() break res=x+y server.send(res) print('server done') if __name__ == '__main__': server,client=Pipe() c1=Process(target=adder,args=((server,client),'c1')) c1.start() server.close() client.send((10,20)) print(client.recv()) client.close() c1.join() print('主进程') #注意:send()和recv()方法使用pickle模块对对象进行序列化。
共享数据 Manager
进程间数据是独立的,可以借助于队列或管道实现通信,二者都是基于消息传递的。
虽然进程间数据独立,但可以通过Manager实现数据共享,事实上Manager的功能远不止于此。
from multiprocessing import Manager,Process,Lock import os def work(d,lock): # with lock: #不加锁而操作共享的数据,肯定会出现数据错乱 d['count']-=1 if __name__ == '__main__': lock=Lock() with Manager() as m: dic=m.dict({'count':100}) p_l=[] for i in range(100): p=Process(target=work,args=(dic,lock)) p_l.append(p) p.start() for p in p_l: p.join() print(dic) #{'count': 94}
信号量(了解) Semaphore
互斥锁同时只允许一个线程更改数据,而Semaphore是同时一定数量的线程更改数据,
信号量与进程池的概念很像,但是要区分开,信号量涉及到加锁的概念。
from multiprocessing import Process,Semaphore import time,random def go_wc(sem,user): sem.acquire() print('%s 占到一个茅坑' %user) time.sleep(random.randint(0,3)) #模拟每个人拉屎速度不一样,0代表有的人蹲下就起来了 sem.release() if __name__ == '__main__': sem=Semaphore(5) p_l=[] for i in range(13): p=Process(target=go_wc,args=(sem,'user%s' %i,)) p.start() p_l.append(p) for i in p_l: i.join() print('============》')
事件(了解) Event
python线程的事件用于主线程控制其他线程的执行,事件主要提供了三个方法:set,wait,clear
clear:将"Flag"设置为False,set:将"Flag"设置为True。
事件处理机制:全局定义了一个"Flag",如果Flag值为False,那么当程序执行event.wait方法时就会阻塞;如果Flag值为True,那么当程序执行event.wait方法时不会阻塞
#_*_coding:utf-8_*_ #!/usr/bin/env python from multiprocessing import Process,Event import time,random def car(e,n): while True: if not e.is_set(): #Flase print('