多进程(二)
1.僵尸进程与孤儿进程
基于unix进程(linux,macOS):
主进程需要等待子进程结束之后,主进程才结束
主进程时刻检测子进程的运行状态,当子进程结束之后一段时间内,将子进程进行回收.
为什么主进程不在子进程结束后马上对其进行回收呢?
主进程与子进程是异步关系,主进程无法马上捕获到子进程什么时候结束
如果子进程结束之后马上在内存中释放资源,主进程就没有办法监测子进程的状态了
unix针对上面的问题,提供了一个机制.
所有的子进程结束之后,立马释放掉文件的操作链接,内存的大部分数据,但是会保留一部分内容:进程号,结束时间,运行状态,等待主进程检测回收.
僵尸进程(有害):
所有的子进程结束之后,立马释放掉文件的操作链接,内存的大部分数据,但是会保留一部分内容:进程号,结束时间,运行状态,等待主进程检测回收.
所有的子进程结束后,在被主进程回收之前,都会进入僵尸进程状态.
危害:
如果父进程不对僵尸进程进行回收(wait/waitpid),产生大量的僵尸进程,这样就会大量占用内存,占用进程pid号.
检测:
from multiprocessing import Process
import time
import os
def task(name):
print(f'{name} is running')
print(f'主进程: {os.getppid()}')
print(f'子进程: {os.getpid()}')
time.sleep(50)
print(f'{name} is gone')
if __name__ == '__main__':
p = Process(target=task,args=('tony',))
p.start()
print('==主开始')
孤儿进程(无害):
父进程由于某种原因结束了,但是子进程还在运行中,这样子进程就变成了孤儿进程,父进程如果结束了,所有的孤儿进程会被init进程回收,init就变成父进程,对子进程进行回收.
2.互斥锁
进程之间数据不共享,但是共享同一套文件系统,而共享带来的就是竞争,竞争带来的结果就是错乱,如何控制,就是加锁处理.
“在编程中，引入了对象互斥锁的概念，来保证共享数据操作的完整性。每个对象都对应于一个可称为 互斥锁 的标记，这个标记用来保证在任一时刻，只能有一个线程访问该对象。”
假设三个人,同时用一个打印机打印
三个进程模拟三个人,输出平台模拟打印机
版本一:
from multiprocessing import Process
import time,random,os
def task1():
print(f'{os.getpid()}开始打印了')
time.sleep(random.randint(1,3))
print(f'{os.getpid()}打印结束了')
def task2():
print(f'{os.getpid()}开始打印了')
time.sleep(random.randint(1,3))
print(f'{os.getpid()}打印结束了')
def task3():
print(f'{os.getpid()}开始打印了')
time.sleep(random.randint(1,3))
print(f'{os.getpid()}打印结束了')
if __name__ == '__main__':
p1 = Process(target = task1)
p2 = Process(target = task2)
p3 = Process(target = task3)
p1.start()
p2.start()
p3.start()
现在是所有的进程都并发的抢占打印机
并发是以效率优先的,但是需求为:顺序优先
多个进程共抢一个资源时,要保证顺序优先:串行,一个一个来
版本二:
from multiprocessing import Process
import time,random,os
def task1(p):
print(f'{p}开始打印了')
time.sleep(random.randint(1,3))
print(f'{p}打印结束了')
def task2(p):
print(f'{p}开始打印了')
time.sleep(random.randint(1,3))
print(f'{p}打印结束了')
def task3(p):
print(f'{p}开始打印了')
time.sleep(random.randint(1,3))
print(f'{p}打印结束了')
if __name__ == '__main__':
p1 = Process(target = task1,args = ('p1',))
p2 = Process(target = task2,args = ('p2',))
p3 = Process(target = task3,args = ('p3',))
p1.start()
p1.join()
p2.start()
p2.join()
p3.start()
p3.join()
利用join解决串行的问题,保证了顺序优先,但是先后顺序是固定的.
这样不合理,在抢夺同一个资源的时候,应该是先到先得,保证公平
版本3:
from multiprocessing import Process
import time,random,os
def task1(p,lock):
lock.acquire()
print(f'{p}开始打印了')
time.sleep(random.randint(1,3))
print(f'{p}打印结束了')
lock.release()
def task2(p,lock):
lock.acquire()
print(f'{p}开始打印了')
time.sleep(random.randint(1,3))
print(f'{p}打印结束了')
lock.release()
def task3(p,lock):
lock.acquire()
print(f'{p}开始打印了')
time.sleep(random.randint(1,3))
print(f'{p}打印结束了')
lock.release()
if __name__ == '__main__':
mutex = Lock()
p1 = Process(target = task1,args = ('p1',mutex))
p2 = Process(target = task2,args = ('p2',mutex))
p3 = Process(target = task3,args = ('p3',mutex))
p1.start()
p2.start()
p3.start()
lock既保证了顺序有限,又保证了公平性
#一把锁不能连续锁两次
#lock.acquire()
#lock.acquire()
#lock.release()
#lock.release()
lock与join的区别:
共同点:都可以把并发变成串行,保证了顺序
不同点:join是人为设定顺序,lock让其争抢顺序,保证了公平性
3.进程之间的通信
进程在内存级别是隔离的,但是文件在磁盘上
1.基于文件的通信
模拟抢票系统:
1.先可以查票,查询余票数量 # 并发
2.进行购买.向服务端发送请求,服务端接收请求,在后端将票数-1,返回到前端 #串行
#文件db.json的内容为：{"count":1}
#一定要用双引号,不然json无法识别
from multiprocessing import Process
import json,time,os,random
def search():
time.sleep(random.randint(1,3)) #模拟网络延迟(查询环节)
with open("db.json","r",encoding = "utf-8") as f:
dic = json.load(f)
print(f'{os.getpid()}查看了票数,剩余{dic["count"]}')
def paid():
with open("db.json","r",encoding = "utf-8") as f:
dic = json.load(f)
if dic["count"] > 0:
dic["count"] -= 1
time.sleep(random.randint(1,3)) #模拟网络延迟(购买环节)
with open("db.json","r",encoding = "utf-8") as f:
json.dump(dic,f)
print(f'{os.getpid()} 购买成功')
def task():
search()
paid()
if __name__ == '__main__':
for i in range(6):
p = Process(target = task)
p.start()
并发运行,效率高,但竞争写同一个文件,数据写入错乱
当多个进程共抢一个数据时,如果要保证数据的安全,必须要串行
要想让购买环节串行执行,我们必须要加锁处理
from multiprocessing import Process
import json,time,os,random
def search():
time.sleep(random.randint(1,3)) #模拟网络延迟(查询环节)
with open("db.json","r",encoding = "utf-8") as f:
dic = json.load(f)
print(f'{os.getpid()}查看了票数,剩余{dic["count"]}')
def paid():
with open("db.json","r",encoding = "utf-8") as f:
dic = json.load(f)
if dic["count"] > 0:
dic["count"] -= 1
time.sleep(random.randint(1,3)) #模拟网络延迟(购买环节)
with open("db.json","r",encoding = "utf-8") as f:
json.dump(dic,f)
print(f'{os.getpid()} 购买成功')
def task(lock):
search()
lock.acquire()
paid()
lock.release()
if __name__ == '__main__':
mutex = Lock()
for i in range(6):
p = Process(target = task,args = (mutex,))
p.start()
当很多进程共抢一个资源(数据)时,要保证顺序(数据的安全),一定要串行
互斥锁:可以公平性的保证顺序以及数据的安全
基于文件的进程之间的通信:
1.效率低下
2.自己加锁麻烦而且很容易出现死锁
2.基于队列的通信
进程彼此之间互相隔离,要实现进程间通信,multiprocessing模块支持两种形式:队列和管道,这两种方式都是使用消息传递的
创建队列的类(底层就是以管道和锁定的方式实现):
Queue(maxsize):创建共享的进程队列,Queue是多进程安全的队列,可以使用Queue实现多进程之间的数据传递
maxsize是队列中允许最大项数,省略则无大小限制
队列:把队列理解成一个容器,可以承载一些数据
队列的特性:先进先出(FIFO,first in first out)
主要方法:
1 q.put方法用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。
2 q.get方法可以从队列读取并且删除一个元素。同样，get方法有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常.
from multiprocessing import Queue
q = Queue(4)
def func():
print('in func')
q.put(1)
q.put('alex')
q.put([1,2,3])
q.put(func)
#q.put(555) 当队列满了时,再put数据会阻塞
print(q.get()) #1
print(q.get()) #'alex'
print(q.get()) #[1,2,3]
f = q.get()
f() # 'in func'
#print(q.get()) 当数据取完时,再get数据也会出现阻塞,知道某一个进程put数据
q.put(55,block = False) #遇到阻塞就会报错
print(q.get(timeout = 3)) #阻塞3秒,3秒后还阻塞则报错