Python多进程

Posted 2020-11-14 戈达尔

进程

什么是进程

正在进行的一个过程或者说一个任务，每个进程在内存中使用的数据彼此是物理级别的隔离

进程的创建分类

1. 系统初始化时创建的进程（在任务管理器中可以看到，这是启动系统后自动创建的进程）
2. 一个进程在运行过程中开启了子进程
3. 用户的交互式请求，而创建一个新进程（当我们打开一个应用的时候）
4. 一个批处理作业的初始化

无论哪一种，新进程的创建都是由一个已经存在的进程执行了一个用于创建进程的系统调用而创建的

进程的三种基本状态

• 就绪

  进程已获得除处理器外的所需资源，等待分配处理器资源

• 运行

  进程已获得处理器资源，正在运行的状态

• 阻塞

  进程遇到某些条件（例如：I/O操作）时，会将该进程暂停并将处理器资源分给其他进程

进程的终止

1. 正常终止：程序执行完毕或用户主动退出
2. 意外终止：程序发生错误，导致程序奔溃/终止
3. 被其他进程终止，例如用户使用任务管理器关闭应用

进程的创建

1. 实例化Process类

   from multiprocessing import Process
   # 进程需要运行的方法
   def pr():
       for i in range(5):
           print(i)
   
   if __name__ == '__main__':
       # 将要执行的方法的方法名传给target
       p = Process(target=pr)
       # 开启进程
       p.start()
       print('主程序运行结束！')
   ###################################
   结果：
   主程序运行结束！
   0
   1
   2
   3
   4

2. 继承Process类

   # 继承Process类
   class MyProcess(Process):
    # 重写run方法（必须）
       def run(self) -> None:
           for i in range(5):
               print(i)
   
   if __name__ == '__main__':
       # 创建实例化对象
       p = MyProcess()
       # 开启进程
       p.start()
       print('主程序运行结束！')
   ###################################
   结果：
   主程序运行结束！
   0
   1
   2
   3
   4

从上面两个例子可以看出来，新创建的进程是独立于主进程的。主进程的代码执行完也不会影响子进程的运行

孤儿进程和僵尸进程

僵尸进程：当使用fork创建子进程的时候，没有对结束的子进程进行回收时，父进程还一直在创建子进程。对于没有被父进程回收的子进程我们称之为 “僵尸进程”

孤儿进程：当子进程的父进程退出，但是子进程还在运行的话，那么子进程就会变成孤儿进程（没有了父进程），这个时候就会有init进程来“收养”这个子进程，并对子进程进行回收资源。

进程间的内存隔离

进程与进程之间是内存隔离的，就如同医院的隔离一样，虽然是在同一个硬件上，但是进程与进程之间的内存占用是分开的。

from multiprocessing import Process
x = 0
def task():
    # 子进程修改的是自己的x，与主进程无关。
    global x 
    x = 50
    print(f'子进程的x：{x}')
if __name__ == '__main__':
    p = Process(target=task)
    p.start()
    print(f'主进程的x:{x}')

进程的属性与方法

1. join

   阻塞主进程，先运行子进程的内容，等子进程运行完成再继续运行主进程的内容

   from multiprocessing import Process
   
   def pr():
       for i in range(5):
           print(i)
   
   if __name__ == '__main__':
   
       p = Process(target=pr)
       p.start()
       # 阻塞主进程
       p.join()
       print('主程序运行结束！')
   ###################################
   结果：
   0
   1
   2
   3
   4
   主程序运行结束！

   如果是创建多个进程，并且在进程开始后调用 join 方法那么多进程就变成了串行。这样子，效率会变得比单纯的调用方法的速度要慢上很多，因为每次进程创建都要分配资源。

2. pid

   查看当前进程的pid号

   from multiprocessing import Process
   
   def pr():
       pass
   
   if __name__ == '__main__':
   
       p = Process(target=pr)
       p.start()
       # 查看进程的pid号
       print(p.pid)   # 7120

   如果要查看当前进程的pid号和当前进程的父进程的pid号的话就要使用os模块的getpid()和getppid()

   from multiprocessing import Process
   import os
   def pr():
       # 查看当前进程pid号
       print(os.getpid())   # 13740
       # 查看当前进程的父进程的pid号
       print(os.getppid())   # 17220
   
   if __name__ == '__main__':
   
       p = Process(target=pr)
       p.start()
       # 查看进程的pid号
       print(p.pid)   # 13740

3. name

   进程的名字

   from multiprocessing import Process
   
   def pr():
        pass
   
   if __name__ == '__main__':
   
       p = Process(target=pr)
       p.start()
       # 查看进程的名字
       print(p.name)   # Process-1

4. is_alive

   判断子进程是否结束

   from multiprocessing import Process,current_process
   import time
   def task():
       time.sleep(1)
   
   if __name__ == '__main__':
       p = Process(target=task)
       p.start()
       print(p.is_alive()) # True
       time.sleep(2)
       print(p.is_alive()) # False

守护进程

• 守护进程会在主进程代码执行结束后就终止
• 守护进程内无法再开启子进程,否则抛出异常

from multiprocessing import Process
import time
def foo():
    print('守护进程开始')
    time.sleep(3)
    print('守护进程开始')


if __name__ == '__main__':
    p1 = Process(target=foo)
    p1.daemon = True
    p1.start()
    time.sleep(1)
    print('主进程')

同步锁

进程之间数据不共享,但是共享同一套文件系统,所以可以访问同一个文件，对同一个文件进行同时读取或写入

当我们要同时读取与写入的时候就会出现问题：我们读取到的是写入前的数据还是写入后的数据，这个我们无法控制，但是我们可以控制同一时间只有一个进程来对这个文件进行操作

from multiprocessing import Lock,Process
import time

def buy(lock):
    #获取钥匙在acquire()和release()之间的代码在同一时间只允许一个进程执行，当一个进程完成后将钥匙归还后，其他的进程才可以获得钥匙。
    lock.acquire()
    with open('access.txt','r',encoding='utf-8') as f:
        count = int(f.read())
    # 模拟网络的延迟
    time.sleep(0.1)
    if count>0:
        print('买到票了')
        count=count-1
        with open('access.txt', 'w', encoding='utf-8') as f:
            f.write(str(count))
    else:
        print('没票了')
    lock.release()  #还锁
if __name__=='__main__':
    #初始化一把锁
    lock=Lock()
    count = 1
    for i in range(3):
        #要把锁当作参数传进去buy()函数中
        p=Process(target=buy,args=(lock,))  
        p.start()

这样即使有多个进程要对文件进行操作，也会由于没有拿到锁而无法对文件进行操作

队列

进程之间数据相互隔离，要实现进程间通信可以使用队列或管道这两种方式。主要使用队列。

from multiprocessing import Process,Queue,current_process

def put(q:Queue):
    for i in range(3):
        print(current_process().name+f'存入{i}')
        q.put(i)

def get(q):
    for i in range(3):
        print(f'{current_process().name}取出'+str(q.get()))

if __name__ == '__main__':
    q = Queue()
    p1 = Process(target=put, args=(q,))
    p1.start()
    p2 = Process(target=get, args=(q,))
    p2.start()

生产者消费者模型

生产者消费者模型具体来讲，就是在一个系统中，存在生产者和消费者两种角色，他们通过内存缓冲区进行通信，生产者生产消费者需要的资料，消费者把资料做成产品，从而消耗掉生产的数据。

from multiprocessing import Process, Queue  # 多进程组件,队列
import time, random


# 生产者方法
def producer(wupin, q):
    for i in range(4):
        time.sleep(random.randint(1, 3))  # 模拟获取数据时间
        f = '生产的%s%s' % (wupin, i)
        print(f)
        q.put(f)  # 添加进队列


# 消费者方法
def consumer(q, name):
    while True:
        food = q.get()  # 如果获取不到，会一直阻塞进程不会结束子进程
        # 当队列中的数据是None的时候结束while循环
        if food is None:
            print('%s获取到一个空' % name)
            break
        f = '33[31m%s消费了%s33[0m' % (name, food)
        print(f)
        time.sleep(random.randint(1, 3))  # 模拟消耗数据时间


if __name__ == '__main__':
    q = Queue()  # 创建队列

    # 模拟生产者 生产数据
    p = Process(target=producer, args=('信件', q))  # 创建进程
    p.start()  # 启动进程
    p1 = Process(target=producer, args=('包裹', q))
    p1.start()

    # 模拟消费者消费数据
    c = Process(target=consumer, args=(q, 'c'))
    c.start()
    c1 = Process(target=consumer, args=(q, 'c1'))
    c1.start()

    p.join()
    p1.join()
    q.put(None)
    q.put(None)