一、multiprocess.process模块
1.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为子进程的名称
2.process类的方法
p.start()：启动进程，并调用该子进程中的p.run()
p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法
p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
p.is_alive():如果p仍然运行，返回True
p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程
3.process类的属性
p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置
p.name:进程的名称
p.pid：进程的pid
p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束(了解即可)
p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）
4.在windows运行时要注意的问题
在Windows操作系统中由于没有fork(linux操作系统中创建进程的机制)，在创建子进程的时候会自动 import 启动它的这个文件，而在 import 的时候又执行了整个文件。因此如果将process()直接写在文件中就会无限递归创建子进程报错。所以必须把创建子进程的部分使用if __name__ ==‘__main__’ 判断保护起来，import 的时候 ，就不会递归运行了。
5.实例代码
1）创建一个子进程，并运行
from multiprocessing import Process def func(name): print('子进程：你好，',name) if __name__ == '__main__': p = Process(target=func,args=('hsr',)) p.start()
2)查看子进程和父进程的进程号
from multiprocessing import Process import os def func(): print('子进程PID:',os.getpid()) if __name__ == '__main__': p = Process(target=func) p.start() print('父进程PID:',os.getpid())
3)join
from multiprocessing import Process import time def func(*args): print('*'*args[0]) time.sleep(5) print('*' * args[1]) if __name__ == '__main__': p = Process(target=func,args=(10,20)) p.start() p.join() #主线程等待p终止 print("-------运行完了-------")
4）开启多个子进程
from multiprocessing import Process import time def func(no,*args): print(str(no)+" :"+'*'*args[0]) time.sleep(5) print(str(no)+" :"+'*'*args[1]) if __name__ == '__main__': p_li = [] for i in range(10): p_li.append(Process(target=func,args=(i,10,20))) for i in p_li: i.start() [i.join() for i in p_li] #让最后的print等子进程都结束了再执行 print('运行完了')
5）实现多进程的另一种方法
#自定义类 继承Process类 #必须实现run方法，run方法就是子进程执行的方法 #如果要参数，则实现自己的init方法，并在其中调用父类的init方法 from multiprocessing import Process import os class MyProcess(Process): def __init__(self,arg1): super().__init__() self.arg1 = arg1 def run(self): print("My Process:",self.pid) print(self.arg1) if __name__ == '__main__': print(os.getpid()) p1 = MyProcess(4) p1.start()
6.进程间的数据隔离
#进程间不会共享数据 from multiprocessing import Process import os def func(): global n n = 0 print('pid:'+str(os.getpid())+" "+str(n)) if __name__ == '__main__': n = 100 p = Process(target=func) p.start() p.join() print('pid:'+str(os.getpid())+" "+str(n))
7.守护进程
守护进程(daemon)是一类在后台运行的特殊进程，用于执行特定的系统任务。很多守护进程在系统引导的时候启动，并且一直运行直到系统关闭。
会随着主进程的结束而结束。
主进程创建守护进程
其一：守护进程会在主进程代码执行结束后就终止
其二：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children
注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止
#守护进程 from multiprocessing import Process import time def func(): while 1: time.sleep(2) print('Good') if __name__ == '__main__': p = Process(target=func) p.daemon = True #设置子进程为守护进程 p.start() i = 10 while i>0: print('Do something') time.sleep(5) i -= 1
二、进程同步
1.进程锁
#加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。
虽然可以用文件共享数据实现进程间通信，但问题是：
1.效率低（共享数据基于文件，而文件是硬盘上的数据）
2.需要自己加锁处理
#因此我们最好找寻一种解决方案能够兼顾：1、效率高（多个进程共享一块内存的数据）2、帮我们处理好锁问题。这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。
队列和管道都是将数据存放于内存中
队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来，
我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往往可以获得更好的可获展性。
#模拟吃50个人吃5个苹果 #使用Lock对象的acquire请求锁，release释放锁 from multiprocessing import Process from multiprocessing import Lock import json def eat(no,lock): lock.acquire() with open('info.json') as f: dic = json.load(f) AppleNum = dic["Apple"] print("苹果个数:" + str(AppleNum)) if AppleNum >0: print("%d 吃了一个苹果" %no) AppleNum -= 1 dic["Apple"] = AppleNum with open('info.json','w') as f: json.dump(dic,f) else: print("%d 没有苹果吃了" %no) lock.release() if __name__ == '__main__': lock = Lock() for i in range(50): Process(target=eat, args=(i,lock)).start()
2.信号量
信号量(Semaphore)，有时被称为信号灯，是在多线程环境下使用的一种设施，是可以用来保证两个或多个关键代码段不被并发调用。在进入一个关键代码段之前，线程必须获取一个信号量；一旦该关键代码段完成了，那么该线程必须释放信号量。其它想进入该关键代码段的线程必须等待直到第一个线程释放信号量。
信号量（semaphore）的数据结构为一个值和一个指针，指针指向等待该信号量的下一个进程。信号量的值与相应资源的使用情况有关。当它的值大于0时，表示当前可用资源的数量；当它的值小于0时，其绝对值表示等待使用该资源的进程个数。注意，信号量的值仅能由PV操作来改变。
对于信号量，可以认为是一个仓库，有两个概念，容量和当前的货物个数。
P操作从仓库拿货，如果仓库中没有货，线程一直等待，直到V操作，往仓库里添加了货物，为了避免P操作一直等待下去，会有一个超时时间。
V操作往仓库送货，如果仓库满了，线程等待，直到有P操作，从仓库中拿走货物，有空的位置。
创建信号量，设置容量，先有V操作，才能P操作。
P操作：货物个数减1，减过之后，货物个数大于等于0，说明已经拿到货物，线程继续。否者线程阻塞。
V操作：货物个数加1，加过之后，货物个数小于等于容量，说明添加成功，线程继续。否者线程阻塞。
信号量：0≤ 信号量≤容量 ，取值 表示当前可以使用的货物；
信号量<0 ， 取值 表示当前等待使用货物的线程；
信号量>容量 ， 信号量-容量 表示当前等待添加货物的线程。
通常，信号量的容量设置很大，可以一直V操作，不会阻塞，但是P操作的时候，很可能阻塞。
当容量为1，也就是互斥，执行流程必定是V操作，P操作，V操作，P操作...
P原语操作的动作是：
（1） sem减1；
（2） 若sem减1后仍大于或等于零，则进程继续执行；
（3） 若sem减1后小于零，则该进程被阻塞后进入与该信号相对应的队列中，然后转进程调度。
V原语操作的动作是：
（1） sem加1；
（2） 若相加结果大于零，则进程继续执行；
（3） 若相加结果小于或等于零，则从该信号的等待队列中唤醒一等待进程，然后再返回原进程继续执行或转进程调度。
from multiprocessing import Process,Semaphore import time import random def grid(i,sem): sem.acquire() print(str(i)+'放入了格子') time.sleep(random.randint(2,6)) print(str(i)+'拿出了格子') sem.release() if __name__ == '__main__': sem = Semaphore(4) for i in range(20): Process(target=grid,args=(i,sem)).start()
3.事件
通过一个信号来控制多个进程同时执行或阻塞
一个事件被创建后，默认是阻塞状态
from multiprocessing import Event if __name__ == "__main__": e = Event() # c创建一个事件 print(e.is_set()) # 查看一个事件的状态 e.set() #将事件的状态改为True e.wait() #根据e.is_set()的值决定是否阻塞 print(1235455) e.clear() #将事件的状态改为False e.wait() print(12323545555)
红绿灯问题
from multiprocessing import Event,Process import time import random def traffic_light(e): while 1: if e.is_set(): e.clear() print('\033[31m[-----------红灯-----------]\033[0m') else: e.set() print('\033[32m[-----------绿灯-----------]\033[0m') time.sleep(2) def car(e,i): if not e.is_set(): print('%s号车在等红灯' %i) e.wait() #阻塞直到状态改变 print('\033[0;32;40m%s号车通过\033[0m' %i) if __name__ == '__main__': e = Event() light = Process(target=traffic_light,args=(e,)) light.start() for i in range(20): time.sleep(random.random()) cars = Process(target=car,args=(e,i)) cars.start()
三、进程间通信
1.队列Queue
Queue([maxsize])
创建共享的进程队列。
参数 ：maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。
底层队列使用管道和锁定实现。
方法
Queue([maxsize])
创建共享的进程队列。maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。底层队列使用管道和锁定实现。另外，还需要运行支持线程以便队列中的数据传输到底层管道中。
Queue的实例q具有以下方法：
q.get( [ block [ ,timeout ] ] )
返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止。block用于控制阻塞行为，默认为True. 如果设置为False，将引发Queue.Empty异常（定义在Queue模块中）。timeout是可选超时时间，用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用，将引发Queue.Empty异常。
q.get_nowait( )
同q.get(False)方法。
q.put(item [, block [,timeout ] ] )
将item放入队列。如果队列已满，此方法将阻塞至有空间可用为止。block控制阻塞行为，默认为True。如果设置为False，将引发Queue.Empty异常（定义在Queue库模块中）。timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。
q.qsize()
返回队列中目前项目的正确数量。此函数的结果并不可靠，因为在返回结果和在稍后程序中使用结果之间，队列中可能添加或删除了项目。在某些系统上，此方法可能引发NotImplementedError异常。
q.empty()
如果调用此方法时 q为空，返回True。如果其他进程或线程正在往队列中添加项目，结果是不可靠的。也就是说，在返回和使用结果之间，队列中可能已经加入新的项目。
q.full()
如果q已满，返回为True. 由于线程的存在，结果也可能是不可靠的（参考q.empty（）方法）。。
简单使用：
from multiprocessing import Process,Queue if __name__ == '__main__': q = Queue(5) #创建队列 for i in range(5): q.put(i) #放进数据 print(q.full()) #q.put(6) 此处阻塞 for i in range(5): print(q.get()) #获取数据 print(q.empty()) #q.get() 此处阻塞
简单的进程间通信：
from multiprocessing import Event,Process,Queue def produce(q): q.put('from produce') def comsume(q): print(q.get()) if __name__ == '__main__': q = Queue(5) #创建队列 pro = Process(target=produce,args=(q,)) pro.start() com = Process(target=comsume, args=(q,)) com.start()
生产者消费者模型
在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。
为什么要使用生产者和消费者模式
在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。
什么是生产者消费者模式
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
1）使用Queue
from multiprocessing import Process,Queue import time import random def producer(name,goods,q): for i in range(10): time.sleep(random.randint(1,4)) print('%s生产了第%s个%s'%(name,i,goods)) q.put('第%s个%s'%(i,goods)) def comsumer(q,name): while 1: goods = q.get() if goods == None:break print('\033[31m%s买了了%s\033[0m' % (name,goods)) time.sleep(random.randint(2,6)) if __name__ == '__main__': q = Queue(10) p = Process(target=producer,args=('HSR','牛奶',q)) p2 = Process(target=producer, args=('TTT', '面包', q)) c = Process(target=comsumer, args=(q,'Lisi')) c2 = Process(target=comsumer, args=(q, 'ZhangSan')) p.start() p2.start() c.start() c2.start() p.join() p2.join() q.put(None) q.put(None)
2）使用JoinableQueue
创建可连接的共享进程队列。这就像是一个Queue对象，但队列允许项目的使用者通知生产者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。
from multiprocessing import Process,JoinableQueue import time import random def producer(name,goods,q): for i in range(10): time.sleep(random.randint(1,4)) print('%s生产了第%s个%s'%(name,i,goods)) q.put('第%s个%s'%(i,goods)) q.join() #阻塞，直到队列中的数据被全部执行完毕 def comsumer(q,name): while 1: goods = q.get() if goods == None:break print('\033[31m%s买了了%s\033[0m' % (name,goods)) time.sleep(random.randint(2,6)) q.task_done() #count - 1 if __name__ == '__main__': q = JoinableQueue(10) p = Process(target=producer,args=('HSR','牛奶',q)) p2 = Process(target=producer, args=('TTT', '面包', q)) c = Process(target=comsumer, args=(q,'Lisi')) c2 = Process(target=comsumer, args=(q, 'ZhangSan')) p.start() p2.start() c.daemon = True #设置为守护进程，主进程结束则子进程结束，而这里的主进程等待生产进程的结束 c2.daemon = True #生产进程又等待消费进程消费完。所以消费者消费完了就会结束进程 c.start() c2.start() p.join() p2.join()
2.管道
multiprocessing.Pipe
from multiprocessing import Pipe,Process def func(conn1,conn2): conn2.close() while 1: try: print(conn1.recv()) except EOFError: conn1.close() break if __name__ == '__main__': conn1, conn2 = Pipe() p1 = Process(target=func,args=(conn1, conn2)) #传给不同进程的conn是不会相互影响的 p1.start() conn1.close() for i in range(20): conn2.send("hi") conn2.close()
使用管道实现生产者消费者模型
#Pipe有数据不安全性 #管道可能出现一端的多个消费者同时取一个数据 #所以可以加上一个进程锁来保证安全性 from multiprocessing import Pipe,Process,Lock import time import random def producer(con,pro,name,goods): con.close() for i in range(8): time.sleep(random.randint(1,3)) print('%s生成了第%s个%s'%(name,i,goods)) pro.send('第%s个%s'%(i,goods)) pro.close() def consumer(con,pro,name,lock): pro.close() while 1: try: lock.acquire() goods = con.recv() lock.release() print('%s喝了%s'%(name,goods)) time.sleep(random.random()) except EOFError: lock.release() #因为最后消费者通过异常来结束进程，所以最后一次的recv后面的lock.release不会执行，所以要在 #这个地方再写一个release() con.close() break if __name__ == '__main__': con, pro = Pipe() lock = Lock() p = Process(target=producer, args=(con,pro,'HSR','牛奶')) c = Process(target=consumer, args=(con, pro, 'TTT',lock)) c2 = Process(target=consumer, args=(con, pro, 'TTT2',lock)) p.start() c.start() c2.start() con.close() pro.close()
3.Manager
multiprocessing.Manager模块
#报 AttributeError: 'ForkAwareLocal' object has no attribute 'connection' 的原因
#运行这段代码时，主进程执行完了，断开了连接，而子进程要连接，此时会报错
#所以可以用join()，让主进程等待子进程的结束
from multiprocessing import Manager,Process def func(dic): dic['count'] -= 1 print(dic) if __name__ == '__main__': m = Manager() 创建一个Manger() dic = m.dict({'count':100}) #变成进程共享的字典 p = Process(target=func, args=(dic,)) p.start() p.join() #等待子进程结束
这里会有进程抢占造成的数据不安全问题，通过加锁解决
from multiprocessing import Manager,Process,Lock 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)
四、进程池
1.为什么要有进程池?
在程序实际处理问题过程中，忙时会有成千上万的任务需要被执行，闲时可能只有零星任务。那么在成千上万个任务需要被执行的时候，我们就需要去创建成千上万个进程么？首先，创建进程需要消耗时间，销毁进程也需要消耗时间。第二即便开启了成千上万的进程，操作系统也不能让他们同时执行，这样反而会影响程序的效率。因此我们不能无限制的根据任务开启或者结束进程。那么我们要怎么做呢？
在这里，要给大家介绍一个进程池的概念，定义一个池子，在里面放上固定数量的进程，有需求来了，就拿一个池中的进程来处理任务，等到处理完毕，进程并不关闭，而是将进程再放回进程池中继续等待任务。如果有很多任务需要执行，池中的进程数量不够，任务就要等待之前的进程执行任务完毕归来，拿到空闲进程才能继续执行。也就是说，池中进程的数量是固定的，那么同一时间最多有固定数量的进程在运行。这样不会增加操作系统的调度难度，还节省了开闭进程的时间，也一定程度上能够实现并发效果。
2.multiprocess.Pool模块
Pool([numprocess [,initializer [, initargs]]]):创建进程池
参数
numprocess:要创建的进程数，如果省略，将默认使用cpu_count()的值
initializer：是每个工作进程启动时要执行的可调用对象，默认为None
initargs：是要传给initializer的参数组
2.方法
p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。 '''需要强调的是：此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数，必须从不同线程调用p.apply()函数或者使用p.apply_async()''' p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。 '''此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。当func的结果变为可用时，将理解传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。''' p.close():关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成 P.jion():等待所有工作进程退出。此方法只能在close（）或teminate()之后调用 方法apply_async()和map_async（）的返回值是AsyncResul的实例obj。实例具有以下方法 obj.get():返回结果，如果有必要则等待结果到达。timeout是可选的。如果在指定时间内还没有到达，将引发一场。如果远程操作中引发了异常，它将在调用此方法时再次被引发。 obj.ready():如果调用完成，返回True obj.successful():如果调用完成且没有引发异常，返回True，如果在结果就绪之前调用此方法，引发异常 obj.wait([timeout]):等待结果变为可用。 obj.terminate()：立即终止所有工作进程，同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收，将自动调用此函数
3.实例
1）简单例子1（使用map）
#map如果要给函数传参数，只能传可迭代对象 from multiprocessing import Pool def func(dic): print(dic) def func2(dic): print(dic+2) if __name__ == '__main__': pool = Pool(5) #进程数，CPU核心数+1 #如果Pool()不传参数，默认是cpu核心数 pool.map(func2,range(100)) #100个任务 #这里自带join效果 pool.map(func, ['hsr','ttt']) # 2个任务
2）简单例子2（使用apply）
from multiprocessing import Pool import os import time def func(n): print('[pid:%s]start id:%s'%(os.getpid(),n)) time.sleep(1.5) print('\033[31m[pid:%s]end id:%s\033[0m'%(os.getpid(),n)) if __name__ == '__main__': pool = Pool(5) for i in range(10): #pool.apply(func,args=(i,)) #同步 pool.apply_async(func,args=(i,)) #异步。与主进程完全异步，需要手动close和join pool.close() # 结束进程池接收任务 pool.join() # 感知进程中的任务都执行结束
3）使用进程池实现多个TCP socket连接
服务端：
import socket from multiprocessing import Pool def func(conn): while 1: conn.send(b'hello') ret = conn.recv(1024).decode('utf-8') if ret == 'q': break print(ret) conn.close() if __name__ == '__main__': sk = socket.socket() sk.bind(('127.0.0.1', 8081)) sk.listen() pool = Pool(5) while 1: conn, addr = sk.accept() pool.apply_async(func,args=(conn,))
客户端：
import socket sk = socket.socket() sk.connect(('127.0.0.1',8081)) ret = sk.recv(1024).decode('utf-8') print(ret) c = input().encode('utf-8') sk.send(c) sk.close()
4.进程池的返回值
from multiprocessing import Pool def func(i): return i**2 if __name__ == '__main__': pool = Pool(5) #使用map的返回值 ret = pool.map(func,range(10)) print(ret) res_l = [] for i in range(10): #同步 # res = pool.apply(func,args=(i,)) #apply的结果就是func的返回值 # print(res) #异步 res = pool.apply_async(func,args=(i,)) #apply_async的结果 #这里如果直接使用res.get()来获取返回值，会阻塞，所以先将其放入列表中，后面再get # print(res.get()) #阻塞等待func的结果 res_l.append(res) for i in res_l: print(i.get())
5.回调函数
需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数
我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果。
from multiprocessing import Pool def func(i): print('in func1') return i**2 def func2(n): print('in func2') print(n) if __name__ == '__main__': pool = Pool(5) pool.apply_async(func, args=(10,), callback=func2) #执行func1，把返回值作为fun2的参数执行func2 #回调函数func2在主进程中zhi'x pool.close() pool.join()
简单例子：
import requests from multiprocessing import Pool def get(url): ret = requests.get(url) if ret.status_code == 200: return ret.content.decode('utf-8'),url def call_back(args): print(args[1] +" "+ str(len(args[0]))) url_lst = [ 'http://www.cnblog.com', 'https://www.baidu.com', 'http://www.sohu.com' ] if __name__ == '__main__': pool = Pool(5) for i in url_lst: pool.apply_async(get,args=(i,),callback=call_back) pool.close() pool.join()