4月25日 python学习总结 互斥锁 IPC通信 和 生产者消费者模型

一、守护进程　　

import random
import time
from multiprocessing import  Process

def task():
    print('name: egon')
    time.sleep(random.randint(0,3))
    print('sex:male')
    time.sleep(random.randint(0, 3))
    print('age:19')


if __name__ == '__main__':
      p=Process(target=task)
      p.daemon()   #将p设置为主进程的守护进程，主进程结束，无论子进程是否正常执行完，都跟主进程一起结束

二、互斥锁

 互斥锁：
强调：必须是lock.acquire()一次，然后 lock.release()释放一次，才能继续lock.acquire()，不能连续的lock.acquire()

 互斥锁vs join的区别：

    大前提：

　　  二者的原理都是一样，都是将并发变成串行，从而保证有序

 

　  区别：

 一、join是按照人为指定的顺序执行，而互斥锁是所以进程平等地竞争，谁先抢到谁执行
 二、 互斥锁可以让一部分代码（修改共享数据的代码）串行，而join只能将代码整体串行

from multiprocessing import Process,Lock
import time,random

mutex=Lock()
# 互斥锁：
#强调：必须是lock.acquire()一次，然后 lock.release()释放一次，才能继续lock.acquire()，不能连续的lock.acquire()

# 互斥锁vs join的区别一：
# 大前提：二者的原理都是一样，都是将并发变成串行，从而保证有序
# 区别：# 互斥锁vs join的区别：
# 大前提：二者的原理都是一样，都是将并发变成串行，从而保证有序
# 区别：一、join是按照人为指定的顺序执行，而互斥锁是所以进程平等地竞争，谁先抢到谁执行
#    二、 互斥锁可以让一部分代码（修改共享数据的代码）串行，而join只能将代码整体串行

def task1(lock):
    lock.acquire()          #抢锁
    print('task1:名字是egon')
    time.sleep(random.randint(1,3))
    print('task1:性别是male')
    time.sleep(random.randint(1,3))
    print('task1:年龄是18')
    lock.release()          #释放锁

def task2(lock):
    lock.acquire()
    print('task2:名字是alex')
    time.sleep(random.randint(1,3))
    print('task2:性别是male')
    time.sleep(random.randint(1,3))
    print('task2:年龄是78')
    lock.release()


def task3(lock):
    lock.acquire()
    print('task3:名字是lxx')
    time.sleep(random.randint(1,3))
    print('task3:性别是female')
    time.sleep(random.randint(1,3))
    print('task3:年龄是30')
    lock.release()



if __name__ == '__main__':
    p1=Process(target=task1,args=(mutex,))
    p2=Process(target=task2,args=(mutex,))
    p3=Process(target=task3,args=(mutex,))

    # p1.start()
    # p1.join()
    # p2.start()
    # p2.join()
    # p3.start()
    # p3.join()

    p1.start()
    p2.start()
    p3.start()

　

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

lock=Lock()
def search():
    print('========%s 查票======' % os.getpid())
    info=json.load(open('test.txt'))
    msg='余票为： %s'%info['count']
    print(msg)

def get(lock):
    lock.acquire()
    print('========%s 抢票======'%os.getpid())
    info = json.load(open('test.txt'))
    time.sleep(random.random())
    if info['count']>0:
        info['count']-=1
        time.sleep(random.random())
        json.dump(info,open('test.txt','w'))
        print('抢票成功')
    else:
        print('票已售完')
    lock.release()

def op(lock):
    search()
    get(lock)

if __name__ == '__main__':
    for i in range(0,50):
        p=Process(target=op,args=(lock,))
        p.start()

　

三、IPC通信（进程之间的的通信控制）　　

进程之间通信必须找到一种介质，该介质必须满足

• 是所有进程共享的
• 必须是内存空间
• 附加：帮我们自动处理好锁的问题

1、队列（推荐使用）     

• 共享的空间
• 是内存空间
• 自动帮我们处理好锁定问题

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

from multiprocessing import Queue

q=Queue(3)    #创建队列，并为队列设置大小 此处为3
q.put('first')
q.put({'second':None})      #可以存放任意类型
q.put('三')

# q.put(4)      #队列存满，第四个存不进，阻塞，等待别的进程取出队列中内容

print(q.get())
print(q.get())
print(q.get())
print(q.get())   # 阻塞 ，队列为空，取不出东西，等待其他进程往队列中存放东西

p.put('first',block=False,timeout=4)
p.get('first',block=False,timeout=4)

#第一个参数 是存放到队列中的数据
#第二个参数block ，是否进入阻塞状态，当队满存值或队空取值时，默认值为True
#第三个参数timeout ，当队满存值或队空取值时,阻塞等待的时间，若超过时间则报错

 

2、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)

　　　　

　　3、管道（不推荐使用）

• 是所有进程共享的
• 是内存空间
• 两头存取数据
• 没有帮我们自动处理好锁的问题

 

四、生产者消费者模型　　

该模型中包含两类重要的角色：

1、生产者：将负责造数据的任务比喻为生产者
2、消费者：接收生产者造出的数据来做进一步的处理，该类人物被比喻成消费者


实现生产者消费者模型三要素

1、生产者
2、消费者
3、队列

什么时候用该模型：

程序中出现明显的两类任何，一类任务是负责生产，另外一类任务是负责处理生产的数据的

该模型的好处：

1、实现了生产者与消费者解耦和
2、平衡了生产力与消费力，即生产者可以一直不停地生产，消费者可以不停地处理，因为二者
不再直接沟通的，而是跟队列沟通

import time
import random
from multiprocessing import Process,Queue

def consumer(name,q):
    while True:
        res=q.get()
        time.sleep(random.randint(1,3))
        print('33[46m消费者===》%s 吃了 %s33[0m' %(name,res))


def producer(name,q,food):
    for i in range(5):
        time.sleep(random.randint(1,2))
        res='%s%s' %(food,i)
        q.put(res)
        print('33[45m生产者者===》%s 生产了 %s33[0m' %(name,res))


if __name__ == '__main__':
    #1、共享的盆
    q=Queue()

    #2、生产者们
    p1=Process(target=producer,args=('egon',q,'包子'))
    p2=Process(target=producer,args=('刘清政',q,'泔水'))
    p3=Process(target=producer,args=('杨军',q,'米饭'))

    #3、消费者们
    c1=Process(target=consumer,args=('alex',q))
    c2=Process(target=consumer,args=('梁书东',q))


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