守护进程,互斥锁,IPC,队列,生产者与消费者模型

Posted 2021-01-30 lizeqian1994

小知识点:在子进程中不能使用input输入!

一.守护进程

守护进程表示一个进程b 守护另一个进程a 当被守护的进程结束后,那么守护进程b也跟着结束了

应用场景:之所以开子进程,是为了帮助主进程完成某个任务,然而,如果主进程认为自己的事情一旦做完了就没有必要使用子进程了,就可以将子进程设置为守护进程

例如:在运行qq的过程,开启一个进程,用于下载文件,然而文件还没有下载完毕,qq就退出了,下载任务也应该跟随qq的退出而结束.

from multiprocessing import Process
import time
def task():
　　print(‘妃子的一生‘)
　　time.sleep(3)
　　print(‘妃子88了‘)
if __name__ == ‘__main__‘:
　　fz = Process(target=task)
　　fz.daemon = True # bool类型 将子进程作为主进程的守护进程,注意:必须在开启子进程前设置!
　　fz.start()
　　print(‘皇帝登基!‘)
　　print(‘当了三年皇帝!‘)
　　time.sleep(1)
　　print(‘皇帝驾崩了!‘)

 

二.互斥锁

当多个进程共享一个数据时,可能会造成数据错乱

1.使用join来让这些进程串行,但是这样就造成了无法并发,并且进程执行的顺序就固定了

2.使用锁 将需要共享的数据增加锁 其他的进程在访问数据时,就必须等待当前进程使用完毕

不使用锁:
多个任务在共享一个数据时
串行效率低 但是不会出问题
并发效率高 但是数据可能错乱
使用锁:
from multiprocessing import Process,Lock
import time,random
def task1(lock):
　　lock.acquire() #是一个阻塞函数,会等到别的子进程释放锁才能继续执行
　　print(‘name1:henry‘)
　　time.sleep(random.randint(1,2))
　　print(‘age1:29‘)
　　time.sleep(random.randint(1, 2))
　　print(‘sex1:man‘)
　　lock.release()
def task2(lock):
　　lock.acquire()
　　print(‘name2:wendy‘)
　　time.sleep(random.randint(1, 2))
　　print(‘age2:24‘)
　　time.sleep(random.randint(1, 2))
　　print(‘sex2:woman‘)
　　lock.release()
def task3(lock):
　　lock.acquire()
　　print(‘name3:irene‘)
　　time.sleep(random.randint(1, 2))
　　print(‘age3:27‘)
　　time.sleep(random.randint(1, 2))
　　print(‘sex3:woman‘)
　　lock.release()
if __name__ == ‘__main__‘:
　　lock = Lock()
　　p1 = Process(target=task1,args=(lock,))
　　p1.start()
　　p2 = Process(target=task2,args=(lock,))
　　p2.start()
　　p3 = Process(target=task3,args=(lock,))
　　p3.start()

 

join 和 锁的区别

1.join中顺序是固定的,不公平

2.join是完全的串行,而锁可以使部分代码串行 其他代码还是并发

互斥锁的使用场景_抢票
from multiprocessing import Process,Lock
import json,time,random
def check_ticket(user):
　　time.sleep(random.randint(1,3))
　　with open(‘ticket.json‘,mode=‘rt‘,encoding=‘utf-8‘) as f:
　　　　dic = json.load(f)
　　　　print(‘用户%s,您好!还剩%s张票!‘%(user,dic[‘count‘]))
def buy_ticket(user):
　　with open(‘ticket.json‘,mode=‘rt‘,encoding=‘utf-8‘) as f:
　　　　dic = json.load(f)
　　　　if dic[‘count‘] > 0:
　　　　　　time.sleep(random.randint(1,3))
　　　　　　dic[‘count‘] -= 1
　　　　　　with open(‘ticket.json‘,mode=‘wt‘,encoding=‘utf-8‘) as f2:
　　　　　　　　json.dump(dic,f2)
　　　　　　　　print(‘用户%s购票成功!‘%user)
def task(user,lock):
　　check_ticket(user)
　　lock.acquire()
　　buy_ticket(user)
　　lock.release()

if __name__ == ‘__main__‘:
　　lock = Lock()
　　for i in range(1,11):
　　　　p = Process(target=task,args=(i,lock))
　　　　p.start()

锁的本质是一个bool类型的数据,在执行代码前会先判断这个值

注意 在使用锁的时候必须保证是同一个锁

锁的实现原理 伪代码
l = False
def task(lock):
　　global l
　　if l == False:
　　　　l = True
　　　　print(‘你好!‘)
　　l = False

 

Lock和Rlock的区别:

from multiprocessing import Lock
lock = Lock()
lock.acquire()
lock.acquire()
print(‘我出来了!‘)
lock.release()
#这个程序会卡死

from multiprocessing import RLock
lock = RLock()
lock.acquire()
lock.acquire()
print(‘我出来了!‘)
lock.release()
#这个程序会打印出结果

Rlock表示可重复锁,特点是可以多次执行acquire
Rlock在执行多次acquire时和普通Lock没有任何区别
而Lock如果多次执行acquire就会被锁死
还要注意的一点是Rlock在一个子进程中执行了几次acquire就要执行几次release,这样才会执行下一个子进程

from multiprocessing import Process,RLock
import time
def task(i,lock):
　　lock.acquire()
　　lock.acquire()
　　print(i)
　　time.sleep(3)
　　lock.release()
　　lock.release()
if __name__ == ‘__main__‘:
　　lock = RLock()
　　p1 = Process(target=task,args=(1,lock))
　　p1.start()
　　p2 = Process(target=task,args=(2,lock))
　　p2.start()

死锁:

指的是锁无法打开导致程序卡死 首先要明确的是一把锁是无法锁死的,正常开发时一把锁足够使用,不需要开多把锁

from multiprocessing import Process,Lock
import time
def task1(l1,l2,i):
　　l1.acquire()
　　print(‘%s抢走了碗!‘%i)
　　time.sleep(1)
　　l2.acquire()
　　print(‘%s抢走了筷子!‘%i)
　　print(‘%s开始吃饭!‘%i)
　　l1.release()
　　l2.release()
def task2(l1,l2,i):
l2.acquire()
　　print(‘%s抢走了筷子!‘%i)
　　l1.acquire()
　　print(‘%s抢走了碗!‘%i)
　　print(‘%s开始吃饭!‘%i)
　　l2.release()
　　l1.releaase()
if __name__ == ‘__main__‘:
　　l1 = Lock()
　　l2 = Lock()
　　p1 = Process(target=task1,args=(l1,l2,‘henry‘))
　　p1.start()
　　p2 = Process(target=task2,args=(l1,l2,‘seulgi‘))
　　p2.start()

 

三.IPC

由于进程之间是相互独立的,所以需要设立方案使得各个进程之间可以相互传递数据

1.使用共享文件,多个进程同时读写一个文件

IO速度慢,传输数据大小不受限制

2.管道 是基于内存的,速度快,但是是单向的,用起来麻烦

3.申请共享内存空间,多个进程可以共享这个内存区域(重点)

速度快,但数据量不能太大

from multiprocessing import Manager,Process
def work(d):
　　d[‘count‘] -= 1
if __name__ == ‘__main__‘:
　　with Manager() as m:
　　　　dic = m.dict({‘count‘:100}) #创建一个共享的字典
　　　　p_l = []
　　　　for i in range(100):
　　　　p = Process(target=work,args=(dic,))
　　　　p_l.append(p)
　　　　p.start()

　　for P in p_l:
　　　　p.join()
　　print(dic)

 

四.队列

队列不只用于进程间通讯,也是一种常见的数据容器

特点:先进先出

优点:即使在多进程下,也可以保证数据不会错乱,因为put和get默认都是阻塞的

堆栈:先进后出

from multiprocessing import Queue
q = Queue(1)#创建一个对列,最多可以存一个数据
q.put(‘henry‘)
q.put(‘wendy‘) #当容器中已经填满的时候,put默认会阻塞
print(‘over‘)

 

from multiprocessing import Queue
q = Queue(1)#创建一个对列,最多可以存一个数据
q.put(‘henry‘)
print(q.get())
print(q.get())#当容器中已经没有数据时,get默认会阻塞
print(‘over‘)

 

from multiprocessing import Queue
q = Queue(1)#创建一个对列,最多可以存一个数据
q.put(‘henry‘)
q.put(‘wendy‘,False) #第二个参数设置为False表示不会阻塞,无论容器满了没满都会强行往里面塞,如果满了抛出异常

 

from multiprocessing import Queue
q = Queue(1)#创建一个对列,最多可以存一个数据
q.put(‘henry‘)
print(q.get())
print(q.get(timeout=3))#timeout 仅用于阻塞时

 

五.生产者与消费者模型

什么是生产者消费者模型?

生产者产生数据的一方

消费者处理数据的一方

 

例如需要做一个爬虫?

1.爬取数据

2.解析数据

爬取和解析都是耗时的操作,如果正常按照顺序来编写代码,会造成解析需要等待爬取,爬取也需要等待解析,这样的话效率就会很低

要提高效率,就是让生产者和消费者解开耦合,自己干自己的

如何实现:

1.将两个任务分配给不同的进程

2.提供一个进程共享的数据容器

from multiprocessing import Process,Queue
import time,random
# 爬数据
def get_data(q):
　　for num in range(1,6):
　　　　print(‘正在爬取第%s个数据...‘%num)
　　　　time.sleep(random.randint(1,2))
　　　　print(‘第%s个数据爬取完成!‘%num)
　　　　#把数据装到队列中
　　　　q.put(‘第%s个数据‘%num)
#解析数据
def parse_data(q):
　　for num in range(1,6):
　　　　#取出数据
　　　　data = q.get()
　　　　print(‘正在解析%s...‘%data)
　　　　time.sleep(random.randint(1,2))
　　　　print(‘%s解析完成!‘%data)
if __name__ == ‘__main__‘:
　　#共享数据容器
　　q = Queue()
　　#生产者进程
　　produce = Process(target=get_data,args=(q,))
　　produce.start()
　　#消费者进程
　　customer = Process(target=parse_data,args=(q,))
　　customer.start()