Python中的生产者消费者模型

Posted lijiangyang

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Python中的生产者消费者模型相关的知识,希望对你有一定的参考价值。

了解知识点:

1、守护进程:

·什么是守护进程:

守护进程其实就是一个‘子进程’,守护即伴随,守护进程会伴随主进程的代码运行完毕后而死掉

·为何用守护进程:

当该子进程内的代码在父进程代码运行完毕后就没有存在的意义了,就应该将进程设置为守护进程,会在父进程代码结束后死掉

 

from multiprocessing import Process

import time,os

def task(name):
print(‘%s is running‘%name)
time.sleep(3)

if __name__ == ‘__main__‘:
p1=Process(target=task,args=(‘守护进程‘,))
p2=Process(target=task,args=(‘正常的子进程‘,))
p1.daemon=True # 一定要放到p.start()之前
p1.start()
p2.start()
print(‘主‘)

守护进程举例

 

技术分享图片

以下是守护进程会迷惑人的范例:

 

#主进程代码运行完毕,守护进程就会结束
from multiprocessing import Process
import time
def foo():
print(123)
time.sleep(1)
print("end123")

def bar():
print(456)
time.sleep(3)
print("end456")

if __name__ == ‘__main__‘:
p1=Process(target=foo)
p2=Process(target=bar)

p1.daemon=True
p1.start()
p2.start()
print("main-------")

‘‘‘
main-------
456
enn456
‘‘‘


‘‘‘
main-------
123
456
enn456
‘‘‘

‘‘‘
123
main-------
456
end456
‘‘‘

2、互斥锁:

互斥锁:可以将要执行任务的部分代码(只涉及到修改共享数据的代码)变成串行

join:是要执行任务的所有代码整体串行

强调:必须是lock.acquire()一次,然后 lock.release()释放一次,才能继续lock.acquire(),不能连续的lock.acquire()。否者程序停在原地。
互斥锁vs join: 
大前提:二者的原理都是一样,都是将并发变成串行,从而保证有序(在多个程序共享一个资源时,为保证有序不乱,需将并发变成串行)
区别一:join是按照人为指定的顺序执行,而互斥锁是所以进程平等地竞争,谁先抢到谁执行
区别二:互斥锁可以让一部分代码(修改共享数据的代码)串行,而join只能将代码整体串行(详见抢票系统)

from multiprocessing import Process,Lock
import json
import os
import time
import random

def check():
time.sleep(1) # 模拟网路延迟
with open(‘db.txt‘,‘rt‘,encoding=‘utf-8‘) as f:
dic=json.load(f)
print(‘%s 查看到剩余票数 [%s]‘ %(os.getpid(),dic[‘count‘]))

def get():
with open(‘db.txt‘,‘rt‘,encoding=‘utf-8‘) as f:
dic=json.load(f)
time.sleep(2)
if dic[‘count‘] > 0:
# 有票
dic[‘count‘]-=1
time.sleep(random.randint(1,3))
with open(‘db.txt‘,‘wt‘,encoding=‘utf-8‘) as f:
json.dump(dic,f)
print(‘%s 购票成功‘ %os.getpid())
else:
print(‘%s 没有余票‘ %os.getpid())


def task(mutex):
# 查票
check()

#购票
mutex.acquire() # 互斥锁不能连续的acquire,必须是release以后才能重新acquire
get()
mutex.release()

 

# with mutex:
# get()

if __name__ == ‘__main__‘:
mutex=Lock()
for i in range(10):
p=Process(target=task,args=(mutex,))
p.start()
# p.join()

模拟抢票

3、IPC通信机制

进程之间通信必须找到一种介质,该介质必须满足
1、是所有进程共享的
2、必须是内存空间
附加:帮我们自动处理好锁的问题
 
a、   from multiprocessing import Manager(共享内存,但要自己解决锁的问题)
b、   IPC中的队列(Queue) 共享,内存,自动处理锁的问题(最常用)
c、   IPC中的管道(Pipe),共享,内存,需自己解决锁的问题

a、用Manager(了解知识点)

from multiprocessing import Process,Manager,Lock
import time

mutex=Lock()

def task(dic,lock):
lock.acquire()
temp=dic[‘num‘]
time.sleep(0.1)
dic[‘num‘]=temp-1
lock.release()

if __name__ == ‘__main__‘:
m=Manager()
dic=m.dict({‘num‘:10})

l=[]
for i in range(10):
p=Process(target=task,args=(dic,mutex))
l.append(p)
p.start()
for p in l:
p.join()
print(dic)

 

b、用队列Queue

1)共享的空间

2)是内存空间

3)自动帮我们处理好锁定问题

from multiprocessing import Queue
q=Queue(3) #设置队列中maxsize个数为三
q.put(‘first‘)
q.put({‘second‘:None})
q.put(‘三‘)
# q.put(4) #阻塞。不报错,程序卡在原地等待队列中清出一个值。默认blok=True
print(q.get())
print(q.get())
print(q.get())

强调:
1、队列用来存成进程之间沟通的消息,数据量不应该过大
2、maxsize的值超过的内存限制就变得毫无意义

了解:
q=Queue(3)
q.put(‘first‘,block=False)
q.put(‘second‘,block=False)
q.put(‘third‘,block=False)
q.put(‘fourth‘,block=False) #报错 queue.Full

q.put(‘first‘,block=True)
q.put(‘second‘,block=True)
q.put(‘third‘,block=True)
q.put(‘fourth‘,block=True,timeout=3) #等待3秒后若还进不去报错。注意timeout不能和block=False连用

q.get(block=False)
q.get(block=False)
q.get(block=False)
q.get(block=False) #报错 queue.Empty

q.get(block=True)
q.get(block=True)
q.get(block=True)
q.get(block=True,timeout=2) #等待2秒后还取不出东西则报错。注意timeout不能和block=False连用

了解

4、生产者与消费者模型

该模型中包含两类重要的角色:
1、生产者:将负责造数据的任务比喻为生产者
2、消费者:接收生产者造出的数据来做进一步的处理,该类人物被比喻成消费者
 
实现生产者消费者模型三要素
1、生产者
2、消费者
3、队列
什么时候用该模型:
程序中出现明显的两类任何,一类任务是负责生产,另外一类任务是负责处理生产的数据的
 
该模型的好处:
1、实现了生产者与消费者解耦和
2、平衡了生产者的生产力与消费者的处理数据的能力

注意:生产者消费者模型是解决问题的思路不是技术。可以用进程和队列来实现,也可以用其他的来实现。

from multiprocessing import JoinableQueue,Process
import time
import os
import random

def producer(name,food,q):
for i in range(3):
res=‘%s%s‘ %(food,i)
time.sleep(random.randint(1,3))
# 往队列里丢
q.put(res)
print(‘33[45m%s 生产了 %s33[0m‘ %(name,res))
# q.put(None)

def consumer(name,q):
while True:
#从队列里取走
res=q.get()
if res is None:break
time.sleep(random.randint(1,3))
print(‘33[46m%s 吃了 %s33[0m‘ %(name,res))
q.task_done()

if __name__ == ‘__main__‘:
q=JoinableQueue()
# 生产者们
p1=Process(target=producer,args=(‘egon‘,‘包子‘,q,))
p2=Process(target=producer,args=(‘杨军‘,‘泔水‘,q,))
p3=Process(target=producer,args=(‘猴老师‘,‘翔‘,q,))
# 消费者们
c1=Process(target=consumer,args=(‘Alex‘,q,))
c2=Process(target=consumer,args=(‘wupeiqidsb‘,q,))
c1.daemon=True
c2.daemon=True

p1.start()
p2.start()
p3.start()
c1.start()
c2.start()

p1.join()
p2.join()
p3.join()
q.join() #等待队列被取干净
# q.join() 结束意味着
# 主进程的代码运行完毕--->(生产者运行完毕)+队列中的数据也被取干净了->消费者没有存在的意义

# print(‘主‘)


























































































































































以上是关于Python中的生产者消费者模型的主要内容,如果未能解决你的问题,请参考以下文章

Python-生成器实现简单的"生产者消费者"模型

python并发编程之多线程守护系列互斥锁生产者消费者模型

进击的Python第九章:paramiko模块线程与进程各种线程锁queue队列生产者消费者模型

Python 生产者消费者模型

python 生产者消费者线程模型

python队列生产者消费者模型