一、关于Pipe(管道)队列就是基于管道的方法,不常用,因为管道中的内容是共享的,数据不安全,而且一个数据取走后,其他人没法接收.
由Pipe方法返回的两个连接对象表示管道的两端。每个连接对象都有send和recv方法（除其他之外）。注意，如果两个进程（或线程）试图同时从管道的同一端读取或写入数据，那么管道中的数据可能会损坏。当然，在使用管道的不同端部的过程中不存在损坏风险。
应该特别注意管道端点的正确管理问题。如果是生产者或消费者中都没有使用管道的某个端点，就应将它关闭。这也说明了为何在生产者中关闭了管道的输出端，在消费者中关闭管道的输入端。
如果忘记执行这些步骤，程序可能在消费者中的recv（）操作上挂起（就是阻塞）。管道是由操作系统进行引用计数的，必须在所有进程中关闭管道的相同一端就会能生成EOFError异常。
因此，在生产者中关闭管道不会有任何效果，除非消费者也关闭了相同的管道端点。
我们的目的就是关闭所有的管道，那么主进程和子进程进行通信的时候，可以给子进程传管道的一端就够了，并且用我们之前学到的，信息发送完之后，再发送一个结束信号None，
那么你收到的消息为None的时候直接结束接收或者说结束循环，就不用每次都关闭各个进程中的管道了。
from multiprocessing import Process,Pipe def f1(conn): from_zhujincheng = conn.recv() print('我是子进程') print('来自主进程的消息:',from_zhujincheng) if __name__ == '__main__': conn1,conn2 = Pipe() #创建一个管道对象,全双工,返回管道的两端,但是一端发送的消息,只能另外一端接收,自己这一端是不能接收的 #可以将一端或者两端发送给其他的进程,那么多个进程之间就可以通过这一个管道进行通信了 p1 = Process(target=f1,args=(conn2,)) p1.start() conn1.send('小宝贝,你在哪') print('我是主进程')
二、关Event() (简称事件)
什么是事件:
用于主线程控制其他线程的执行，事件主要提供了三个方法 set、wait、clear。
事件的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。
只应用于2个进程之间,因为进程多了,该机制方法不成立
方法介绍:
e.clear：将“Flag”设置为False
e.set：将“Flag”设置为True
from multiprocessing import Process,Event e = Event() #创建事件对象,这个对象的初识状态为False print('e的状态是:',e.is_set()) print('进程运行到这里了') e.set() #将e的状态改为True print('e的状态是:',e.is_set()) e.clear() #将e的状态改为False e.wait() #e这个事件对象如果值为False,就在我加wait的地方等待 print('进程过了wait')
三、Semaphore()方法(简称信号量)
什么是信号量: 互斥锁同时只允许一个线程更改数据，而信号量Semaphore是同时允许一定数量的线程更改数据 。信号量同步机制适用于访问像服务器这样的有限资源,默认是1个进程
参数介绍:
s = Semaphore(4) #实例化信号量,为4个
s.acquire() #加锁 (百度翻译为:获得)
s.release() #解锁 (百度翻译为:释放,解除)
信号量的使用:
from multiprocessing import Process,Semaphore import time,random def go_ktv(sem,user): sem.acquire() #加锁,信号量没有with方法 print('%s 占到一间ktv小屋' %user) time.sleep(random.randint(0,3)) #模拟每个人在ktv中待的时间不同 sem.release() #解锁 if __name__ == '__main__': sem=Semaphore(4) #实例化信号量 p_l=[] for i in range(13): #参数1是子进程,参2传进去信号量和正常所需参数 p=Process(target=go_ktv,args=(sem,'user%s' %i,)) p.start() #下指令,开启进程 p_l.append(p) #将这个进程名放进列表,方便后面统一阻塞 for i in p_l: i.join() print('============》')
四、关于进程池
什么是进程池:
定义一个池子，在里面放上固定数量的进程，有需求来了，就拿一个池中的进程来处理任务，等到处理完毕，进程并不关闭，而是将进程再放回进程池中继续等待任务。
如果有很多任务需要执行，池中的进程数量不够，任务就要等待之前的进程执行任务完毕归来，拿到空闲进程才能继续执行。也就是说，池中进程的数量是固定的,那么同一时间最多有固定数量的进程在运行。
特点是:这样不会增加操作系统的调度难度，还节省了开闭进程的时间，也一定程度上能够实现并发效果
参数介绍:
p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。 p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。 p.close():关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成 P.jion():等待所有工作进程退出。此方法只能在close（）或teminate()之后调用
p.map(obj,inter):快速生成n项(可迭代项)任务,交给进程池,属于异步提交方式,自带close,join属性
使用进程池运算和创建进程运算,时间的对比
import time from multiprocessing import Process,Pool def f1(n): for i in range(5): n = n+i if __name__ == '__main__': #使用进程池的执行代码所需时间 s_time = time.time() pool=Pool(4) #有100个任务,但是只有4个进程能执行. pool.map(f1,range(100)) #参数1为开启进程的对象,参数2必须是可迭代的,属于异步提交方式 e_time = time.time() dif_time = e_time - s_time #使用多进程执行代码所需的时间 p_s_t = time.time() #起始时间 p_list = [] for i in range(100): p = Process(target=f1,args=(i,)) p.start() p_list.append(p) [pp.join() for pp in p_list] p_e_t = time.time() p_dif_t = p_e_t - p_s_t print("进程池的时间:",dif_time) print("多进程的执行时间:",p_dif_t) #打印结果为:每次结果都不一致,但是进程池用比多进程用的时间短,是一定的. #进程池的时间: 0.1326451301574707 #多进程的执行时间: 2.274912118911743
res = pool.apply(obj,args=(i,)) 进程池的同步提交方式,必须等任务执行结束才能给进程池提交下一个任务,可以直接拿到返回结果res
#进程池的同步异步方法 import time from multiprocessing import Pool def f1(n): time.sleep(0.5) return n*n if __name__ == '__main__': pool = Pool(4) for i in range(10): res = pool.apply(f1,args=(i,)) #同步方法,就是把它该成串行,可以接收返回值 print(res)
pool.apply_async(obj,args(i,)) #进程池的异步提交方式
import time from multiprocessing import Process,Pool def f1(n): time.sleep(0.5) return n*n if __name__ == '__main__': pool = Pool(4) res_list = [] for i in range(10): res = pool.apply_async(f1,args=(i,)) print(res) #这里打印的就是一堆结果集(一堆内存地址) res_list.append(res) for i in res_list: print(i.get()) #get() 获取返回值 print("主进程结束") #解释:为什么打印出来的结果是有序的? # 因为放进列表里面的时候,是有顺序的,你get()的时候,用的列表里面的数据 # 为什么结果是四个四个的出来? # 因为进程池的大小为4
一次性获取到所有进程执行完后的数据关于进程池的回调函数
什么是回调函数:进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数.我们可以把耗时间（阻塞）的任务放到进程池中，
然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果
回调函数是在主进程里面执行的.
回调函数在写的时候注意一点，回调函数的形参执行有一个，如果你的执行函数有多个返回值，那么也可以被回调函数的这一个形参接收，接收的是一个元祖，包含着你执行函数的所有返回值。
import os from multiprocessing import Pool,Process def f1(n): print("参数值为",n) return n*n def call_back_full(s): print("call进程id",os.getpid()) print("回调函数的结果:",s) if __name__ == '__main__': pool = Pool(4) res = pool.apply_async(f1,args=(5,),callback=call_back_full) #callback 译为,回调 pool.close() #如果想让主程序等待进程池,必须把进程池锁住,pool.close(),在pool.join不然会报错 pool.join() print("主进程id",os.getpid())