Python并发编程之进程通信

Posted 2022-11-02 michealjy

'''
进程间的通信
'''

"""
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也没有存在的价值了，不需要继续阻塞在进程中影响主进程了，
应该随着主进程的结束而结束，所以设置成守护进程就可以了
'''