标签:targe 进程 创建 线程 RoCE 生产者 python 模块 并发编程
'''
进程间的通信
'''
"""
multiprocessing模块支持进程间通信的两种主要形式:管道和队列
都是基于消息传递实现的,
"""
from multiprocessing import Queue
q = Queue(3)
#put, get, put_nowait, get_nowait, full, empty
q.put(1)
q.put(2)
q.put(3)
# q.put(1)#队列已经满了,再加程序就会一直停在这里,等待数据被别人取走,不取走就一直停在这
## q.get(1)#可以先取出一个,然后再加就可以了
## q.put(1)
# try:
# q.put_nowait(1) #使用这个绑定方法,队列满了不会阻塞,但是会因为队列满了而报错。
# except:
# print('队列已经满了') #加了try之后,不阻塞,但是消息会丢
print(q.full())#查看队列是否满了
print(q.get())
print(q.get())
print(q.get())
# q.get(1)#同put 方法一样,队列空了继续取就会出现阻塞。
#和上面方法类似
try:
q.get_nowait()
except:
print('队列已经空了')
print(q.empty())
# 子进程数据给父进程
import time
from multiprocessing import Process, Queue
def f(q):
q.put([time.asctime(), 'from Eva', 'hello'])
if __name__ == '__main__':
q = Queue() #创建一个Queue对象
p = Process(target=f, args=(q, )) #创建一个进程
p.start()
print(q.get())
p.join()
#
# #批量生产数据放入队列再批量获取结果
import os
import time
import multiprocessing
#向queue中输入数据的函数
def inputQ(queue):
info = str(os.getpid()) + '(put): ' + str(time.asctime())
queue.put(info)
#向queue中输出数据的函数
def outputQ(queue):
info = queue.get()
print(f'{str(os.getpid())} (get): {info}')
#Main
if __name__ == '__main__':
multiprocessing.freeze_support()
record1 = [] #store input processes
record2 = [] #store output processes
queue = multiprocessing.Queue(3)
#输入进程
for i in range(10):
process = multiprocessing.Process(target=inputQ, args=(queue, ))
process.start()
record1.append(process)
#输出进程
for i in range(10):
process = multiprocessing.Process(target=outputQ, args = (queue, ))
process.start()
record2.append(process)
for p in record1:
p.join()
for p in record2:
p.join()
import os
import time
import multiprocessing
#向队列中输入数据
def inputQ(queue):
info = str(os.getpid()) + '(put):' + str(time.asctime())
queue.put(info)
#向队列中输出数据
def outputQ(queue):
info = queue.get()
print(f'{str(os.getpid())} (get: ) {info}')
if __name__ == '__main__':
multiprocessing.freeze_support()
p1_list = []
p2_list = []
q = multiprocessing.Queue(3)
for i in range(10):
p = multiprocessing.Process(target=inputQ, args=(q, ))
p.start()
p1_list.append(p)
for i in range(10):
p = multiprocessing.Process(target=outputQ, args=(q, ))
p.start()
p2_list.append(p)
for k in p1_list:
k.join()
for j in p2_list:
j.join()
#基于队列实现生产者和消费者模型(生产者生产出来放到阻塞队列里,消费者直接从阻塞队列中取需要的东西)
from multiprocessing import Process, Queue #导入进程和队列模块
import time, random,os#导入时间、随机数和os模块
def consume(q):#定义消费者函数
while True: #循环消费
res = q.get() #从队列中取东西
time.sleep(random.randint(1, 3)) #随机睡几秒
print(f'{str(os.getpid())} 吃 {res}') #打印出来
def producer(q): #定义生产者函数
for i in range(10): #生产10个包子
time.sleep(random.randint(1, 3)) #随机睡几秒
res = f'包子{i}' #生产包子标记下来
q.put(res) #把包子放到阻塞队列里面
print(f'生产了{(os.getpid(), res)}') #打印东西
if __name__ == '__main__':
q = Queue() #阻塞队列
#生产者们:即厨师们
p1 = Process(target=producer, args=(q, ))
#消费者们:即吃货们
c1 = Process(target=consume, args=(q, ))
#开始
p1.start()
c1.start()
print('主')
#生产者和消费者(改良版)
'''
上面的版本主进程永远不会结束,原因是:生产者p在生产完后就结束了,
但是消费者c在取空了q之后,一直处于死循环卡在q.get这一步
解决方法是生产者生产完成之后,网队列中组发一个结束信号,这样消费者在接收到结束
信号后就可以break出死循环
注意:结束信号None,不一定要由生产者发,主进程里面同样可以发,但主进程需要
等生产者结束后才应该发送该信号
'''
import os, random, time
from multiprocessing import Process, Queue
#定义消费者函数
def consume(queue):
while True:
res = queue.get()
if res is None:break #收到结束信号则结束
time.sleep(random.randint(1, 3))
print(f'{str(os.getpid())} 吃了 {res}')
#定义生产者函数
def producer(queue):
for i in range(10):
time.sleep(random.randint(1, 3))
res = f'包子{i}'
queue.put(res)
print(f'{os.getpid()}生产了{res}')
# queue.put(None) #生产者发送结束信号
if __name__ == '__main__':
q = Queue(3)
#生产者
p = Process(target=producer, args=(q, ))
#消费者
q1 = Process(target=consume, args=(q, ))
p.start()
q1.start()
p.join()
q.put(None) #主进程里面发送结束信号
q1.join() #可加可不加
print('主线程')
# 多个消费者就需要发送多个None
from multiprocessing import Process,Queue
import time,random,os
def consumer(q):
while True:
res=q.get()
if res is None:break #收到结束信号则结束
time.sleep(random.randint(1,3))
print('%s 吃 %s' %(os.getpid(),res))
def producer(name,q):
for i in range(2):
time.sleep(random.randint(1,3))
res='%s%s' %(name,i)
q.put(res)
print('生产了 %s %s' %(os.getpid(),res))
if __name__ == '__main__':
q=Queue()
#生产者们:即厨师们
p1=Process(target=producer,args=('包子',q))
p2=Process(target=producer,args=('骨头',q))
p3=Process(target=producer,args=('泔水',q))
#消费者们:即吃货们
c1=Process(target=consumer,args=(q,))
c2=Process(target=consumer,args=(q,))
#开始
p1.start()
p2.start()
p3.start()
c1.start()
# 必须保证生产者全部生产完毕,才应该发送结束信号
p1.join()
p2.join()
p3.join()
# 有几个消费者就应该发送几次结束信号None
q.put(None)
q.put(None) #发送结束信号
print('主')
#JoinableQueue队列实现消费者与生产者模型
from multiprocessing import Process,JoinableQueue
import time,random,os
def consumer(q):
while True:
res=q.get()
time.sleep(random.randint(1,3))
print('%s 吃 %s' %(os.getpid(),res))
#向q.join()发送一次信号,证明一个数据已经被取走了
q.task_done()
def producer(name,q):
for i in range(2):
time.sleep(random.randint(1,3))
res='%s%s' %(name,i)
q.put(res)
print('生产了 %s %s' %(os.getpid(),res))
q.join() #生产完毕,使用此方法进行阻塞,直到队列中所有项目均被处理
if __name__ == '__main__':
q=JoinableQueue()
#生产者们:即厨师们
p1=Process(target=producer,args=('包子',q))
p2=Process(target=producer,args=('骨头',q))
p3=Process(target=producer,args=('泔水',q))
#消费者们:即吃货们
c1=Process(target=consumer,args=(q,))
c2=Process(target=consumer,args=(q,))
c1.daemon=True #设置成守护进程
c2.daemon=True #设置成守护进程
#开始
p_1 = [p1, p2, p3, c1, c2]
for p in p_1:
p.start()
# 必须保证生产者全部生产完毕,才应该发送结束信号
p1.join()
p2.join()
p3.join()
print('主')
'''
主进程等--->p1, p2, p3等--->c1, c2
p1, p2, p3结束了,证明c1, c2肯定全部收完了p1, p2, p3发到队列的数据
因而c1, c2也没有存在的价值了,不需要继续阻塞在进程中影响主进程了,
应该随着主进程的结束而结束,所以设置成守护进程就可以了
'''
标签:targe 进程 创建 线程 RoCE 生产者 python 模块 并发编程
原文地址:https://www.cnblogs.com/michealjy/p/11523507.html