进程之间的通信:
1. 对列
2. 管道
进程之间的数据共享
进程池
对列 先进先出FIFO ---维护秩序的时候用的比较多 买票 秒杀
from queue import Queue
q = Queue(5) #括号里面的数字是设置队列的长度
print(q.qsize()) #查看队列的当前长度
q.put(12) #向队列中添加内容
q.get() #取出内容
当队列中满了或者空了,再去填值或者取值就会阻塞.
print(q.full()) #查看队列是否满了
print(q.empty()) #查看队列是否空了
q.put_nowait(4) #如果队列满了继续放值,则会报错.
q.get_nowait() #如果队列空了继续取值,则会报错.
栈 : 先进后出 ----算法
实例 : 三级菜单
计算文件夹的总大小
进程之间的通信 --- IPC
队列
from multiprocessing import Queue
q = Queue()
在多进程中 q.qsize() q.empty() q.full()是不准的
q.empty()
q.full()
q.put_nowait()
q.put()
q.get()
q.get_nowait()
q.qsize()
主进程放 子进程取
from multiprocessing import Queue , Process
def con(q):
print(q.get())
if __name__ == "__main__":
q = Queue()
p = Process(target=con,args=(q,))
p.start()
q.put(123)
子进程放 另一个子进程取
from multiprocessing import Queue , Process
def con(q):
print(q.get())
def pro(q):
q.put(123)
if __name__ == "__main__":
q = Queue()
p = Process(target=con,args=(q,))
p.start()
c = Process(target=pro,args=(q,))
c.start()
生产者消费者模型 -- 解决创造(生产)数据和处理(消费)数据的效率不平衡问题
把创造数据和处理数据放在不同的进程中 , 根据他们的效率来调整进程的个数
生产数据快 消费数据慢 内存空间的浪费
消费数据快 生产数据慢 效率低下
import time,random
from multiprocessing import Process,Queue
def consumer(q,name):
while True:
food = q.get()
if food == "stop" : break
print("%s 消费了 %s"%(name,food))
time.sleep(random.random())
def product(q,name,food,n = 10):
for i in range(n):
time.sleep(random.random())
fd = food + str(i)
print("%s 生产了%s"%(name,fd))
q.put(fd)
if __name__ == "__main__":
q = Queue()
c = Process(target=consumer,args=(q,"alex""))
c.start()
p = Process(target=product,args=(q,"egon","美女")
p.start()
p.join()
q.put("stop")
让consumer停下来的方法
1.在所有生产者结束生产之后向队列中放入一个结束符
2.有几个consumer就向队列中放几个结束符
3.在消费之消费的过程中,接收到结束符,就结束消费的进程
import time,random
from multiprocessing import JoinableQueue , Process
def consumer(q,name):
while True:
food = q.get()
print("%s 吃了%s"%(name,food))
time.sleep(random.random())
q.task_done()
def product(q,name,food,n= 10):
for i in range(10):
time.sleep(random.random())
fd = food + str(i)
print("%s 生产了%s"%(name,fd))
q.put(fd)
q.join(()
if __name__ == "__main__":
q = JoinableQueue()
p = Process(target=product,args=(q,"egon","美女"))
p.start()
c = Process(target=consumer,args=(q,"alex"))
c.daemon = True
c.start()
只有multiprocessing中的对列,才能帮助实现IPC
永远不可能出现数据不安全的情况,多个进程不会同时取走同一个数据
由于对列先进先出的特点 + 进程通信的功能 + 数据进程安全,经常用它来完成进城之间的通信
生产者消费者模型:
生产者和消费者的效率平衡的问题
内存的控制 -- 对列的长度限制
让消费者自动停下来
JoinableQueue
在消费数据的时候 : task_done
在生产端主进程 : join
管道 : 是IPC通信的一种机制,队列就是基于管道来完成通信的,但是管道是原生的通信方式,在进程之间会产生数据不安全的情况,需要自己手动加锁来处理.管道在数据传输过程中,还涉及到一个端口管理这个需要我们在代码中处理,才能使代码更完善.
队列就是基于管道实现的
对列 : 数据是安全的
管道 : 数据是不安全的
对列 = 管道 + 锁
from multiprocessing import Pipe
left ,right = Pipe()
left.send("hello")
print(right.recv())
from multiprocessing import Pipe,Process
def consumer(left,right):
left.close()
while 1:
try:
print(right.recv())
except EOFError:
break
if __name__ == "__main__":
left,right = Queue()
p = Process(target=consumer,args=(left,right))
p.start()
right.close()
for i in range(5):
left.send("hello")
left.close()
EOF异常的出发:
在这一个进程中如果不在用这个端点了,应该close
这样在recv的时候,如果其他端点都被关闭了,就能够知道不会再有新的消息传进来,此时就不会在这里阻塞等待,而是抛出一个EOFError
close并不是关闭了整个管道,而是修改了操作系统对管道端点的应用计数的处理
from multiprocessing Process,Pipe
def consumer(p,name):
left,right = p
left.close()
while 1:
try:
ba = right.recv()
except EOFError:
break
def product(p,seq=10):
left,right = p
right.close()
for i in range(seq):
left.send("hello")
if __name__ == "__main__":
left,right=Pipe()
for i in range(5):
p = Process(target=consumer,args=((left,right),"nana"))
p.start()
for i in range(5):
c = Process(target=product,args=((left,right),))
c.start()
left.close()
right.close()
6.数据共享 : Manager模块
from multiprocessing import Manager,Process,Lock
def work(d,lock):
with lock:
d["count"] -= 1
if __name__ == "__main__":
lock = Lock()
m = Manager() #共享数据
dic = m.dict({"count":100}) #Manager调用字典
pl = []
for i in range(100):
p = Process(target=work,args=(dic,lock)) #所有子进程共享字典
pl.append(p)
p.start()
for i in pl:
i.join()
print(dic)
7.进程池的初识:
操作系统开启进程慢,几个CPU就能同时运行几个进程,进程的个数不是无限开启的
进程池 : 如果必须用多个进程,且是高计算性,没有IO型的程序,希望并行,最充分的使用CPU.
import os,time
from multiprocessing import Pool
def func(i):
time.sleep(0.1)
print(os.getpid(),i)
if __name__== "__main__":
p = Pool(5)
for i in range(20):
p.apply_async(func,args=(i,))
p.close()
p.join()
8.总结:
IPC通信:
队列 : 管道 + 锁
管道 : 是队列的底层
数据共享 : 进程就是数据隔离的
Manager模块
数据类型都能够进行数据共享
一部分都是不加锁,不支持数据进程安全
不安全的解决办法 是加锁
进程池
进程不能无限开,会给操作系统调度增加负担
且真正能被同时执行的进程最多也就和CPU个数相同等
进程的开启和销毁都要消耗资源和时间
进程池 和信号量 :
进程池 :
进程池里有几个进程,就起几个.不管多少任务,池子里进程的个数是固定的.
开启进程和关闭进程这些事都是需要固定的时间开销 , 不会产生额外的时间开销 ; 且进程池中的进程数控制的好,那么操作系统的压力也小.
信号量 :
有多少个任务就起多少个进程/线程
可以帮助减少操作系统切换负担 , 但是并不能帮助减少进/线程开启和关闭的时间