多进程补充

多进程补充

僵尸进程和孤儿进程

基于unix环境(linux,macOS)

• 主进程须要等待子进程结束以后,主进程才结束

  主进程时刻监测子进程的运行状态,当子进程结束以后,一段时间以内,将子进程进行回收.

• 为何主进程不在子进程结束后立刻对其回收呢?

  1. 主进程与子进程是异步关系.主进程没法立刻捕获子进程何时结束.
  2. 若是子进程结束以后立刻再内存中释放资源,主进程就没有办法监测子进程的状态了.

• unix针对于上面的问题,提供了一个机制.

  全部的子进程结束以后,立马会释放掉文件的操做连接,内存的大部分数据,可是会保留一些内容: 进程号,结束时间,运行状态,等待主进程监测,回收.

• 僵尸进程: 全部的子进程结束以后,在被主进程回收以前,都会进入僵尸进程状态.

• 僵尸进程有无危害???

  若是父进程不对僵尸进程进行回收(wait/waitpid),产生大量的僵尸进程,这样就会占用内存,占用进程pid号.

• 孤儿进程:

  父进程因为某种缘由结束了,可是你的子进程还在运行中,这样你的这些子进程就成了孤儿进程.你的父进程若是结束了,你的全部的孤儿进程就会被init进程的回收,init就变成了你的父进程,对你进行回收.

• 僵尸进程如何解决???

  父进程产生了大量子进程,可是不回收,这样就会造成大量的僵尸进程,解决方式就是直接杀死父进程,将全部的僵尸进程变成孤儿进程进程,由init进行回收.

互斥锁

互斥锁就是在保证子进程串行的同时,也保证了子进程执行顺序的随机性,以及数据的安全性

# 三个同事 同时用一个打印机打印内容.
# 三个进程模拟三个同事, 输出平台模拟打印机.

版本一:
from multiprocessing import Process
import time
import random
import 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
import random
import 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',))

    p2.start()
    p2.join()
    p1.start()
    p1.join()
    p3.start()
    p3.join()
# 咱们利用join 解决串行的问题,保证了顺序优先,可是这个谁先谁后是固定的.
# 这样不合理. 你在争抢同一个资源的时候,应该是先到先得,保证公平.

版本3:

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

def task1(p,lock):
    '''
    一把锁不能连续锁两次
    lock.acquire()
    lock.acquire()
    lock.release()
    lock.release()
    '''
    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))

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

版本四：
from multiprocessing import Process
from multiprocessing import Lock
import time
import random
import sys
import os
def task(name,Lock):
    Lock.acquire()
    print(f"{name} is running")
    time.sleep(random.randint(1,4))
    print(f"{name} is gone")
    Lock.release()
if __name__ == '__main__':
    mutex = Lock()
    for i in range(3):
        p = Process(target=getattr(sys.modules[__name__],'task'),args=(f"p{i}",mutex))
        p.start()

lock与join的区别.

共同点: 均可以把并发变成串行, 保证了顺序.

不一样点: join人为设定顺序,lock让其争抢顺序,保证了公平性.

# 当不少进程共强一个资源(数据)时, 你要保证顺序(数据的安全),必定要串行.
# 互斥锁: 能够公平性的保证顺序以及数据的安全.

# 基于文件的进程之间的通讯:
    # 效率低.
    # 本身加锁麻烦并且很容易出现死锁.

进程之间的通讯

进程在内存级别是隔离的,可是文件在磁盘上,

基于文件通讯

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


def search():
    time.sleep(random.randint(1,3))  # 模拟网络延迟(查询环节)
    with open('ticket.json',encoding='utf-8') as f1:
        dic = json.load(f1)
        print(f'{os.getpid()} 查看了票数,剩余{dic["count"]}')


def paid():
    with open('ticket.json', encoding='utf-8') as f1:
        dic = json.load(f1)
    if dic['count'] > 0:
        dic['count'] -= 1
        time.sleep(random.randint(1,3))  # 模拟网络延迟(购买环节)
        with open('ticket.json', encoding='utf-8',mode='w') as f1:
            json.dump(dic,f1)
        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
from multiprocessing import Lock
import json
import time
import os
import random


def search():
    time.sleep(random.randint(1,3))  # 模拟网络延迟(查询环节)
    with open('ticket.json',encoding='utf-8') as f1:
        dic = json.load(f1)
        print(f'{os.getpid()} 查看了票数,剩余{dic["count"]}')


def paid():
    with open('ticket.json', encoding='utf-8') as f1:

        dic = json.load(f1)
    if dic['count'] > 0:
        dic['count'] -= 1
        time.sleep(random.randint(1,3))  # 模拟网络延迟(购买环节)
        with open('ticket.json', encoding='utf-8',mode='w') as f1:
            json.dump(dic,f1)
        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()

基于队列通讯

from multiprocessing import Process
from multiprocessing import Queue
import random
import time
import os
def check(q):
    time.sleep(random.randint(1,3))
    num = q.qsize()
    print(f"{os.getpid()}查票，剩余{num}")
def paid(q):
    time.sleep(random.randint(1,3))
    try:
        q.get(block = False)
        if q.qsize()>=0:
            print(f"{os.getpid()}购买成功，剩余{q.qsize()}")
        else:
            print('没票了')
    except Exception:
        print('没票了')
def task(q):
    check(q)
    paid(q)
if __name__ == '__main__':
    q = Queue()
    for i in range(3):
        q.put(1)
    for i in range(10):
        p = Process(target=task,args=(q,))
        p.start()