多进程实操

目录

• multiprocessing模块
• Process类
• Process类的使用
• 创建并开启子进程的两种方式
• 进程直接的内存空间是隔离的
• start方法
• join方法
• 守护进程（daemon）

---

multiprocessing模块

跨平台的进程创建模块。Python中的多线程无法利用多核优势，如果想要充分地使用多核CPU的资源，在Python中大部分情况需要使用多进程。

multipeocessing模块用来开启子进程，并在子进程中执行我们定制的任务（比如函数），该模块与多线程模块threading的编程接口类似。

multiprocessing模块的功能众多：支持子进程、通信、数据共享、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件。

与线程不同，进程没有任何共享的状态，进程修改的数据，改动仅限于该进程内。

---

Process类

创建进程的类。在 multiprocessing 中，通过创建一个 Process 对象然后调用它的 start() 方法来生成进程。 Process 和 threading.Thread API 相同。

Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，
表示一个子进程中的任务（尚未启动）

强调：
1. 需要使用关键字的方式来指定参数
2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号

Process参数：

参数	含义
group	未使用，值始终为None
target	表示调用对象，即子进程要执行的任务
args	表示调用对象的位置参数元组，例如，args=（‘李白’，）
kwargs	表示调用对象的关键字参数字典，例如，kwargs={'name':'libai','age'=18}
name	子进程的名称

Process对象的方法：

方法	作用
p.start()	给系统发信号，启动进程，并调用子进程中的p.run()
p.run()	进程启动时执行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法。
p.terminate()	强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁，那么也不会释放，进而导致死锁。
p.is_alive()	判断进程是否存活。
p.join([timeout])	默认timeout为None，将join（）阻塞，使主进程处于等待状态，直到调用者也就是子进程p的终止。 timeout是一个正数，表示最多阻塞多少秒。如果进程终止或方法超时，则join（）返回None。 一个进程可以被 join 多次。进程无法join自身，因为这会导致死锁。 p.join只能join住start开启的进程，而不能join住run开启的进程。不能在start之前执行join。
p.close()	关闭Process对象，释放与之关联的所有资源。如果底层进程仍在运行，则会引发ValueError。

Process对象的属性：

属性	含义
p.daemon	默认为False，如果为True，代表p为后台运行的守护进程,当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置。
p.name	进程的名称
p.pid	进程的pid
p.exitcode	进程在运行时为None，如果为-N，表示被信号N结束
p.authkey	进程的身份验证键，默认是由os.urandom()随机生成的32位字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同身份验证键时才能成功。

---

Process类的使用

注意：在Windows中，Process（）必须放到if __name__ == '__main__':判断的代码体内。而linux操作系统是把内存中的变量直接拷贝，不再去执行一遍。

注：子进程在创建时，所有的数据都是从父进程复制的。

由于Windows没有fork，多处理模块启动一个新的Python进程并导入调用模块。如果在导入时调用Process（），则会引发无限多个新进程（或者直到您的计算机耗尽资源），会抛出RuntimeError的异常：

RuntimeError: 
        An attempt has been made to start a new process before the
        current process has finished its bootstrapping phase.

        This probably means that you are not using fork to start your
        child processes and you have forgotten to use the proper idiom
        in the main module:

            if __name__ == '__main__':
                freeze_support()
                ...

        The "freeze_support()" line can be omitted if the program
        is not going to be frozen to produce an executable.

在子进程中打印下__name__会看到：

from multiprocessing import Process

def f():
    print(__name__)

if __name__ == '__main__':
    p = Process(target=f)
    p.start()

__mp_main__

---

创建并开启子进程的两种方式

方式一：

from multiprocessing import Process

def f(name):
    print(f'hello,{name}')

if __name__ == '__main__':
    p1 = Process(target=f, args=('libai',))
    p2 = Process(target=f, args=('libai',))
    p3 = Process(target=f, args=('libai',))

    p1.start()  # 给操作系统发信号,开启一个进程来执行提交的函数.
    p2.start()
    p3.start()

方式二：

自定义类继承Process类，在里面重写run方法。与方式一相比，方式二可以自定义run方法的执行。

from multiprocessing import Process


class NewProcess(Process):  # run方法的参数不能改变,所以要在初始化时给函数传参
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self) -> None:  # run的返回值应为None
        print(f'hello {self.name}')


if __name__ == '__main__':
    p1 = NewProcess('libai')
    p2 = NewProcess('libai')
    p3 = NewProcess('libai')

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

hello libai
hello libai
hello libai

---

进程直接的内存空间是隔离的

from multiprocessing import Process

x = 10

def f():
    global x
    x = 22
    print(f'子进程x: {x}')

if __name__ == '__main__':
    p = Process(target=f)
    p.start()
    print(f'主进程x: {x}')

主进程x: 10
子进程x: 22

---

start方法

import os
from multiprocessing import Process

def f():
    print('子进程执行')
    print(f'子进程pid: {os.getpid()}')
    print(f'父进程pid: {os.getppid()}')

if __name__ == '__main__':
    p = Process(target=f)
    p.start() 
    print(f'子进程pid: {p.pid}')

start()是发起系统调用启动进程，这个过程需要分配资源，比较慢，所以函数代码会迟一些执行。

子进程pid: 8180
子进程执行
子进程pid: 8180
父进程pid: 7388

---

join方法

import os
import time
from multiprocessing import Process

def f():
    print('子进程执行')
    time.sleep(2)
    print(f'子进程pid: {os.getpid()}')
    print(f'父进程pid: {os.getppid()}')

if __name__ == '__main__':
    p = Process(target=f)
    p.start()
    p.join()
    print(f'主进程中查看子进程pid: {p.pid}')

join会等到子进程运行完毕，所以会先执行函数f：

子进程执行
子进程pid: 6228
父进程pid: 1284
主进程中查看子进程pid: 6228

异步提交：异步提交时，join的顺序不会对总耗时产生影响。

import time
from multiprocessing import Process


class NewProcess(Process):  # run方法的参数不能改变,所以要在初始化时给函数传参
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self) -> None:  # run的返回值应为None
        print(f'hello {self.name}')
        time.sleep(1)


if __name__ == '__main__':
    p1 = NewProcess('libai')
    p2 = NewProcess('libai')
    p3 = NewProcess('libai')
    start = time.time()
    p1.start()
    p2.start()
    p3.start()
    
    p1.join()
    p2.join()
    p3.join()

    end = time.time()
    print(f'进程总耗时:{end - start}')

多个任务异步提交，所以总耗时为进程的执行时间加CPU切换的时间：

hello libai
hello libai
hello libai
进程总耗时:1.2050690650939941

同步提交：

import time
from multiprocessing import Process


class NewProcess(Process):  # run方法的参数不能改变,所以要在初始化时给函数传参
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self) -> None:  # run的返回值应为None
        print(f'hello {self.name}')
        time.sleep(1)


if __name__ == '__main__':
    p1 = NewProcess('libai')
    p2 = NewProcess('libai')
    p3 = NewProcess('libai')
    start = time.time()
    p1.start()
    p1.join()

    p2.start()
    p2.join()

    p3.start()
    p3.join()

    end = time.time()
    print(f'进程总耗时:{end - start}')

这样修改主程序会依次等待子进程执行完毕后才会启动下一个子进程，所以是同步提交：

hello libai
hello libai
hello libai
进程总耗时:3.46219801902771

---

守护进程（daemon）

1、守护进程是守护主程序的代码（而不是生存周期）的进程；正常情况下主进程会等到子进程执行结束才会结束。

2、守护进程内无法再开启子进程，否则抛出异常：AssertionError: daemonic processes are not allowed to have children

import time
from multiprocessing import Process
def f():
    from multiprocessing import Process
    def foo():
        print('这是foo')

    p2 = Process(target=foo)
    p2.start()

if __name__ == '__main__':
    p1 = Process(target=f)
    p1.daemon = True # 要在start之前设置
    p1.start()
    time.sleep(1)

结果抛异常了：

AssertionError: daemonic processes are not allowed to have children

举例：

import time
from multiprocessing import Process

def f1():
    print(123)
    time.sleep(1)
    print('f1_end')
    
def f2():
    print(456)
    time.sleep(3)
    print('f2_end')
    
if __name__ == '__main__':
    p1 = Process(target=f1)
    p2 = Process(target=f2)
    
    p1.daemon = True
    p1.start()
    p2.start()
    print('main......')

如果start较慢，那么结果通常为：

main......  # 这行执行完后守护进程就结束,所以并不会执行f1
456
f2_end

---