一、守护进程
-
守护进程:表示 一个进程b守护另一个进程a,当被守护的进程a结束后,那么b也跟着结束了。 就像皇帝驾崩,妃子殉葬
-
应用场景:之所以开启子进程,是为了帮助进程完成某个任务,然而如果主进程认为 自己的事情一旦做完就没有必要使用子进程了,就可以将子进程设置为守护进程。例如在运行qq的过程开启了一个进程用于下载文件,然而文件还没有下完,qq就退出了,下载任务也应该跟随qq 的退出而结束
示范:
import time
from multiprocessing import Process
def task():
print("妃子的一生")
time.sleep(5)
print("妃子凉了")
if __name__ == '__main__':
fz = Process(target=task)
fz.daemon = True # 将子进程作为主进程的守护进程,要注意必须在开启子进程之前设置!
fz.start()
print("皇帝登基了")
time.sleep(2)
print("当了十年皇帝..")
print("皇帝驾崩")
二、互斥锁
- 为什么用锁:多个进程共享一个数据时,可能会造成数据错乱,使用join来让这些进程串行,但是这将造成无法并发,并且进程执行任务的顺序就固定了。使用锁将需要共享的数据加锁,其他进程在访问数据时,就必须等待当前进程使用完毕
- 锁的本质就是一个bool类型的数据,在执行代码前会先判断这个值,注意在使用锁时 必须保证锁是同一个
from multiprocessing import Process,Lock
import random
import time
def task1(lock):
lock.acquire() # 是一个阻塞的函数,会等到别的进程释放锁才能继续执行
print("1my name is:bgon")
time.sleep(random.randint(1,2))
print("1my age is:78")
time.sleep(random.randint(1,2))
print("1my sex is:femal")
lock.release()
def task2(lock):
lock.acquire()
print("2my name is:blex")
time.sleep(random.randint(1,2))
print("2my age is:68")
time.sleep(random.randint(1,2))
print("2my sex is:femal")
lock.release()
if __name__ == '__main__':
lock = Lock()
p1 = Process(target=task1,args=(lock,))
p1.start()
p2 = Process(target=task2,args=(lock,))
p2.start()
- RLock 表示可重入锁 特点是 可以多次执行acquire,Rlock在执行多次acquire时和普通Lock没有任何区别,如果在多进程中使用Rlock,并且一个进程a,执行了多次acquire,其他进程b要想获得这个锁,需要进程a把锁解开,并且锁了几次就要解几次,普通锁如果多次执行acquire将会锁死
from multiprocessing import Process,Lock,RLock
import time
def task(i,lock):
lock.acquire()
lock.acquire()
print(i)
time.sleep(3)
lock.release()
lock.release()
#第一个过来睡一秒,第二个过来了睡一秒,第一个打印1,第二个打印2
if __name__ == '__main__':
lock = RLock()
p1 = Process(target=task,args=(1,lock))
p1.start()
p2 = Process(target=task, args=(2,lock))
p2.start()
三、互斥锁的使用场景_抢票
#在本地建立一个ticket.json文件内容为{"count": 0}
import json
from multiprocessing import Process,Lock
import time
import random
# 查看剩余票数
def check_ticket(usr):
time.sleep(random.randint(1,3))
with open("ticket.json","r",encoding="utf-8") as f:
dic = json.load(f)
print("%s查看 剩余票数:%s" % (usr,dic["count"]))
def buy_ticket(usr):
with open("ticket.json","r",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", "w", encoding="utf-8") as f2:
json.dump(dic,f2)
print("%s 购票成功!" % usr)
def task(usr,lock):
check_ticket(usr)
lock.acquire()
buy_ticket(usr)
lock.release()
if __name__ == '__main__':
lock = Lock()
for i in range(10):
p = Process(target=task,args=("用户%s" % i,lock))
p.start()
#p.join() #只有第一个人购买完毕,别人才能买这是不公平的
总结:
join和锁的区别:
- join中顺序是固定的不公平
- join是完全串行,而锁可以使部分代码串行,其他代码还是并发
四、死锁
-
死锁指的是锁无法打开了,导致程序卡死,一把锁是不会锁死的,正常开发时,一把锁足够使用,不要开多把锁
死锁示范:
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("吃饭..")
l1.release()
l2.release()
def task2(l1,l2,i):
l2.acquire()
print("筷子被%s抢走了" % i)
l1.acquire()
print("盘子被%s抢走了" % i)
print("吃饭..")
l1.release()
l2.release()
if __name__ == '__main__':
l1 = Lock()
l2 = Lock()
Process(target=task1,args=(l1,l2,1)).start()
Process(target=task2,args=(l1,l2,2)).start()
#筷子被2抢走了
#盘子被1抢走了
五、IPC
- 什么是IPC:进程间通讯,由于进程之间是相互独立的,所以需要对应解决方案,能够使得进程之间可以相互传递数据
- 使用共享文件,多个进程同时读写同一个文件
- 管道是基于内存的,速度快,但是是单向的,用起来麻烦(了解)
- 申请共享内存空间,多个进程可以共享这个内存区域(重点)
from multiprocessing import Manager,Process,Lock
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): #创建100个进程
p=Process(target=work,args=(dic,))
p_l.append(p)
p.start()
for p in p_l:
p.join()
print(dic) #执行结果为{'count': 0}
六、队列
- 队列:不只用于进程间通讯,也是一种常见的数据容器
- 特点:先进先出
- 优点:可以保证数据不会错乱,即使在多进程下,因为其put和get默认都是阻塞的,对比堆栈刚好相反(后进先出)
from multiprocessing import Queue
q = Queue(1) # 创建一个队列,最多可以存一个数据
q.put("张三")
print(q.get())
from multiprocessing import Queue
q = Queue(1)
q.put("张三")
q.put("李四")
print(q.get()) # put默认会阻塞,当容器中已经装满了
print(q.get()) # get默认会阻塞,当容器中已经装满了
from multiprocessing import Queue
q = Queue(1) # 创建一个队列,最多可以存一个数据
q.put("张三")
print(q.get())
print(q.get(timeout=3)) # timeout 仅用于阻塞时
七、生产者消费者模型
-
什么是生产者消费者模型:生成者产生数据的一方,消费者处理数据的一方
例如:需要做一个爬虫
扫描二维码关注公众号,回复: 4940462 查看本文章- 爬取数据
- 解析数据
爬取和解析都是耗时操作,如果正常按照顺序来编写代码,将造成解析需要等待爬取, 爬去取也需要等待解析,这样效率是很低的。要提高效率,就是一个原则,让生产者和消费解开耦合,自己干自己的
如何实现:
- 将两个任务分别分配给不同进程
- 提供一个进程共享的数据容器
import random
from multiprocessing import Process,Queue
import time
# 爬取数据
def get_data(q):
for num in range(5):
print("正在爬取第%s个数据" % num)
time.sleep(random.randint(1,2))
print("第%s个数据 爬取完成" % num)
# 把数据装到队列中
q.put("第%s个数据" % num)
# 解析数据
def parse_data(q):
for num in range(5):
# 取出数据
data = q.get()
print("正在解析%s" % data)
time.sleep(random.randint(1, 2))
print("%s 解析完成" % data)
if __name__ == '__main__':
# 共享数据容器
q = Queue(5)
#生产者进程
produce = Process(target=get_data,args=(q,))
produce.start()
#消费者进程
customer = Process(target=parse_data,args=(q,))
customer.start()