进程之间的通信（ＩＰＣ）队列和管道
一.队列 基于管道实现 管道 + 锁 数据安全
(一).队列
队列遵循先进先出原则(FIFO)
多用于维护秩序,买票,秒杀
队列的所有方法:
put()(给队列里添加数据),put_nowait(),
get()(从队列中获取数据),get_nowait(),
相同点:有值的时候取值
区别:get()没有值时会阻塞
get_nowait() 没有值时会报错
full()(返回布尔值),empty()(返回bool值),
qsize()(队列大小)
示例：
from queue import Queue
q=Queue()
print(q.qsize()) #
q.put()
q.put()
q.put()
q.put()
# q.put_nowait() #已知队列长度,队列未放满数据执行此方法会报错
# 报错信息: TypeError: put_nowait() missing required positional argument: 'item'
q.put()
print(q.qsize()) #
print(q.full()) #True #队列满了返回True.
print(q.get())
print(q.get())
print(q.get())
print(q.get())
print(q.get())
print(q.empty()) #True 队列里无值返回True
print(q.qsize()) # 以上五步取出队列中的数据,此时队列长度是0.
# q.get() #队列里无值,用此方法程序阻塞不返回任何东西
# q.get_nowait() #队列里无值,用此方法程序会报错,报错信息:queue.Empt
补充知识:
栈 ：先进后出
应用：三级菜单 计算文件夹的总大小
（二）．进程中的队列
进程中的队列方法:
empty(), full()在多进程中是不可靠
put(),put_nowait()
get(),get_nowait()
示例:
from multiprocessing import Queue
q=Queue()
q.put()
q.put()
q.put()
print(q.qsize()) #
# q.put_nowait() # 未给队列传参数或报错TypeError: put_nowait() missing required positional argument: 'obj'
print(q.get()) #
print(q.get()) #
print(q.get()) #
print(q.qsize()) #
q.get() #无值不返回
q.get_nowait() #无值会报错queue.Empty
主进程放 子进程取
from multiprocessing import Queue,Process
def con(q): #定义一个con函数 传一个参数 q 是创建的队列对象
print(q.get()) #获取队列数据,如果没有数据则一直阻塞直到有数据为止
if __name__=="__main__": #在这一判断条件下面创建子进程
q=Queue() #创建一个队列对象
p=Process(target=con,args=(q,)) #创建一个子进程子进程对象为con,并传一个参数q.
p.start() #申请开启子进程
q.put() #给队列中添加数据
子进程放,另一个子进程取
from multiprocessing import Queue,Process
def con(q): #定义一个con函数 传一个参数 q 是创建的队列对象
print(q.get()) #获取队列数据,如果没有数据则一直阻塞直到有数据为止
if __name__=="__main__": #在这一判断条件下面创建子进程
q=Queue() #创建一个队列对象
p=Process(target=con,args=(q,)) #创建一个子进程子进程对象为con,并传一个参数q.
p.start() #申请开启子进程
q.put() #给队列中添加数据
生产者消费者模型
--解决创造(生产)数据和处理(消费)数据的效率不平衡问题
把创造数据和处理数据放在不同的进程中
根据他们的效率来调整进程的个数
生产数据快 消费数据慢 内存空间的浪费
生产数据慢 消费数据快 效率低下
import time
import random
from multiprocessing import Queue,Process
def consumer(q,name):#创建一个函数消费者
while True: #无限循环
food=q.get() # 获取队列中数据
if food=="stop":break #.判断获取的数据是否是结束符,如果是则停止获取数据(消费食物)
print("%s吃了%s"% (name,food)) #.提示消费者吃了生产者生产的食物
time.sleep(random.random()) #.消费者消费食物需要一定的时间 5_8 重复一次(消费者进程进行了两次) def producer(q,name,food,n=):#创建一个函数生产者
for i in range(n): #.生产者生产食物的数量计数器
time.sleep(random.random()) #.生产食物需要消耗一定的时间
fd=food+str(i) #11给食物标号
print("%s生产了%s"% (name,fd)) #12提示生产者生产的食物
q.put(fd) #13把食物放进队列 -18重复一遍 ,生产者进程进行了两次 if __name__=="__main__":#在此判断条件下创建一个子进程
q=Queue() #创建一个队列
c=Process(target=consumer,args=(q,"alex"))#创建一个消费者子进程,,传两个参数
c.start() #申请开启子进程
c1=Process(target=consumer,args=(q,"alex")) #再创建一个消费者子进程,传两个参数
c1.start()#申请开启子进程
p=Process(target=producer,args=(q,"太白","食物")) #创建一个生产者子进程,传三个参数
p.start()#申请开启子进程
p1 = Process(target=producer, args=(q, "egon", "甜点"))#再创建一个生产者子进程,传三个参数
p1.start()#申请开启子进程
p.join() #第一个生产者生产结束再执行下一步
p1.join() #第二个生产者生产结束再执行下一步
q.put("stop") #全部生产者生产结束以后 再发送结束符
q.put("stop") #有几个消费者就发送几个结束符
让consumer 停下来的方法
在所有生产者结束生产之后,向队列中放入一个结束符
有几个消费者就向队列中放入几个结束符
在消费者消费的过程中,接收到结束符,就结束消费的进程 生产者消费者模型 ---JoinableQueue版
import time
import random
from multiprocessing import JoinableQueue,Process
def consumer(q,name):#创建消费者函数
while True: #.无限循环
food=q.get() #.获取队列中的食物
print("%s吃了%s" %(name,food))#8提示消费者吃了食物
time.sleep(random.random()) #.消费者消费食物需要时间
q.task_done() #.每消费完一个食物就给生产者发送一次信息 - 以此类推 def producer(q,name,food,n=):#创建生产者函数
for i in range(): #.无限循环 - 以此类推
time.sleep(random.random()) #.生产食物需要耗费时间
fd=food+str(i) #.生产食物名+序号
print("%s生产了%s"%(name,fd))#.提示生产者生产了食物
q.put(fd) #.将生产的食物添加到队列中
q.join() #等消费者全部消费结束之后才结束阻塞# if __name__=="__main__":#在此条件下创建一个子进程
q=JoinableQueue() #创建一个队列(JoinableQueue)
c=Process(target=consumer,args=(q,"alex")) #创建一个消费者进程两个参数
c.daemon=True #设置守护进程
c.start() #申请开启消费者进程
c1=Process(target=consumer,args=(q,"alex")) #在创建一个消费者进程,传两个参数
c1.daemon=True #设置守护进程
c1.start() #申请开启另一个消费者进程
p=Process(target=producer,args=(q,"太白","食物")) #创建一个生产者进程 传三个参数
p.start() #申请开启一个消费者进程
p1=Process(target=producer,args=(q,"egon",'甜点')) #再创建一个生产者进程,传三个参数
p1.start() #申请开启另一个生产者进程
p.join() #结束第一个生产者进程后再执行后面的代码
p1.join() #结束第二个生产者进程后结束主程序 (守护程序直接结束)(消费者程序结束)
总结 :
只有multiprocessing中的队列才能帮助你实现IPC(进程之间的通信)
永远不可能出现数据不安全的情况,多个进程不会同时取走同一个数据
由于先进先出的特点 + 进程通信的功能 + 数据进程安全,经常用它来完成进程之间的通
生产者消费者模型
生产者和消费者的效率平衡的问题
内存的控制--队列的长度限制
让消费者自动停下来
JoinableQueue
在消费数据的时候 task_done
在生产端\主进程 join
流程:
消费者消费完毕
生产者结束阻塞
主进程代码结束
守护进程结束=>消费者结束
二.管道 数据不安全
示例:
from multiprocessing import Pipe
left,right=Pipe()
left.send("888")
print(right.recv())
管道第一版:
from multiprocessing import Pipe,Process
def consumer(left,right):
left.close() # 一个进程中,如果接收时其中一端关闭, 另一端开启在无限循环接收中会报错,
while True: #左右两端都开启则可以无限循环接收,不会报错
try:
print(right.recv()) #接收信息
except EOFError:
break
if __name__=="__main__":
left,right=Pipe() #创建一个管道
p=Process(target=consumer,args=(left,right)) #创建一个consumer子进程,传两个参数
p.start() #申请开启一个子进程
right.close() #关闭右管道口
for i in range(): #计数器
left.send("hello") #左管道口发送信息(发送十次)
left.close() #所有信息发完关闭右管道口
EOF异常的触发:
在这一个进程中,如果不再用这个端点了,应该close.
这一在recv的时候,如果其他端点都被关闭了,就能够知道不会再有新的消息传进来
此时就不会在这里阻塞等待,而是抛出一个EOFError
close并不是关闭了整个管道,而是修改了操作系统对管道端点的引用
一个进程发另一个进程收
from multiprocessing import Process,Pipe
def consumer(p,name):
produce,consume=p
produce.close()
while True:
try:
food=consume.recv()
print("%s收到%s"%(name,food))
except EOFError:
break def producer(p,sep=):
produce,consume=p
consume.close()
for i in range(sep):
produce.send(i) if __name__=="__main__":
produce,consume=Pipe()
for i in range():
c=Process(target=consumer,args=((produce,consume),"c1"))
c.start()
for i in range():
p=Process(target=producer,args=((produce,consume),))
p.start()
producer((produce,consume))
produce.close()
consume.close()
进程之间的数据共享
from multiprocessing import Manager,Process,Lock
def work(d,lock):#定义一个work函数 传两个参数
with lock:
d["count"]-=#字典的count对应的值减一 if __name__=="__main__":
lock=Lock()
m=Manager()
dic=m.dict({"count":})
p_l=[]
for i in range():
p=Process(target=work,args=(dic,lock)) #创建一个work子进程 传两个参数dic,lock,一共创建一百个
p_l.append(p) #每创建一个进程就添加进列表中
p.start() #每创建一个就申请开启一个
for p in p_l: #循环遍历列表中每一个进程
p.join() #直到所有进程都结束再执行后面的代码
print(dic) # 最后打印字典