生产者消费者模型

import time
import random
from multiprocessing import JoinableQueue,Process
# join  阻塞
def consumer(q,name):
    while True:
        food = q.get()
        print(‘%s 吃了 %s‘%(name,food))
        time.sleep(random.random())
        q.task_done()

def producer(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)
    q.join()

if __name__ == ‘__main__‘:
    q = JoinableQueue()
    c1 = Process(target=consumer,args=(q,‘alex‘))
    c1.daemon = True
    c1.start()
    c2 = Process(target=consumer, args=(q, ‘alex‘))
    c2.daemon = True
    c2.start()
    p1 = Process(target=producer,args=(q,‘太白‘,‘泔水‘))
    p1.start()
    p2 = Process(target=producer, args=(q, ‘egon‘, ‘鱼刺‘))
    p2.start()
    p1.join()
    p2.join()

from multiprocessing import Process,JoinableQueue
import time,random,os
def consumer(q):
    while True:
        res=q.get()
        time.sleep(random.randint(1,3))
        print(‘\033[45m%s 吃 %s\033[0m‘ %(os.getpid(),res))
        q.task_done() #向q.join()发送一次信号,证明一个数据已经被取走了

def producer(name,q):
    for i in range(10):
        time.sleep(random.randint(1,3))
        res=‘%s%s‘ %(name,i)
        q.put(res)
        print(‘\033[44m%s 生产了 %s\033[0m‘ %(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_l=[p1,p2,p3,c1,c2]
    for p in p_l:
        p.start()

    p1.join()
    p2.join()
    p3.join()
    print(‘主‘) 
    
    #主进程等--->p1,p2,p3等---->c1,c2
    #p1,p2,p3结束了,证明c1,c2肯定全都收完了p1,p2,p3发到队列的数据
    #因而c1,c2也没有存在的价值了,不需要继续阻塞在进程中影响主进程了。应该随着主进程的结束而结束,所以设置成守护进程就可以了。

from multiprocessing import Process,Pipe

def consumer(p,name):
    produce, consume=p
    produce.close()
    while True:
        try:
            baozi=consume.recv()
            print(‘%s 收到包子:%s‘ %(name,baozi))
        except EOFError:
            break

def producer(seq,p):
    produce, consume=p
    consume.close()
    for i in seq:
        produce.send(i)

if __name__ == ‘__main__‘:
    produce,consume=Pipe()

    c1=Process(target=consumer,args=((produce,consume),‘c1‘))
    c1.start()


    seq=(i for i in range(10))
    producer(seq,(produce,consume))

    produce.close()
    consume.close()

    c1.join()
    print(‘主进程‘)

多个pipe实现生产者消费者模型,数据不安全