Python多进程详解

文章目录

• 1. 多进程
• 2. 创建进程
• • 2.1 直接创建
  • 2.2 继承创建
• 3. 守护进程和join()方法
• 4. 进程锁
• 5. 进程通信
• • 5.1 Queue
  • 5.2 Pipe
• 6. 进程数据共享
• • 6.1 Value
  • 6.2 Array
• 7. 进程池
• 参考文章

前面的多线程文章已经讲过了，Python中的多线程实际上是一种虚假的多线程，在大多时候甚至起到的效果是让运行时间更加的漫长，至少在Python中使用的价值不高，就比如写爬虫的时候只有多进程才能做到同时利用多个CPU内核，使得程序程序运行时间大大缩短，下面我们就来讲讲Python中的多进程。

1. 多进程

进程是指一个程序在给定数据集合上的一次执行过程，是系统进行资源分配和运行调用的独立单位。可以简单地理解为操作系统中正在执行的程序。也就说，每个应用程序都有一个自己的进程。

每一个进程启动时都会最先产生一个线程，即主线程。然后主线程会再创建其他的子线程。

Python中的进程类使用的是 multiprocessing 模块，该模块的API大部分复制了 threading 模块的API，该模块的常用方法如下。

import multiprocessingprint('子进程的列表：{}'.format(multiprocessing.active_children()))print('电脑的CPU数量：{}'.format(multiprocessing.cpu_count()))print('现在运行的进程：{}'.format(multiprocessing.current_process()))

输出如下：

子进程的列表：[]
电脑的CPU数量：12
现在运行的进程：

2. 创建进程

创建线程的方法有两种，一种是直接使用 multiprocessing 模块里面的类来进行创建，一种是继承 multiprocessing 模块的类写一个类来对线程进行创建。

2.1 直接创建

我们可以通过直接从 multiprocessing.Process 继承创建一个新的子类，并实例化后调用 start() 方法启动新进程，即相当于它调用了进程的 run() 方法。

该方法的参数如下：

Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

target 指要创建的进程的方法名，name 指给此进程命名，命名后可以调用 multiprocessing.current_process().name 方法输出该进程的名字， args/kwargs 指 target 指向的方法需要传递的参数，必须是元组形式，如果只有一个参数，需要以添加逗号。

假如我们创建两个进程，一个每隔两秒对传入的数加2，一个每隔1秒对传入的数加1，代码示例如下：

import multiprocessingfrom multiprocessing import Processimport timedef process1(num):while True:num += 2print('{} is running >> {}'.format(multiprocessing.current_process().name, num))time.sleep(2)if __name__ == '__main__':# 设置进程new_pro = Process(target=process1, name='Add2', args=(100,))# 进程开始new_pro.start()n = 0while True:n += 1print('{} is running >> {}'.format(multiprocessing.current_process().name, n))time.sleep(1)

输出结果如下：

MainProcess is running >> 1
Add2 is running >> 102
MainProcess is running >> 2
MainProcess is running >> 3
Add2 is running >> 104
MainProcess is running >> 4
MainProcess is running >> 5
Add2 is running >> 106

注意使用多进程的时候需要特别注意，必须要有 if __name__ == '__main__':， 该语句下的代码相当于主进程，没有该语句会报错。

简单解释下上述内容，由于Python运行过程中，新创建进程后，进程会导入正在运行的文件，即如果没有 if __name__ == '__main__':，代码在运行到 new_pro 时，新的进程会重新读入代码，新进程认为其是要再次运行的代码，这是子进程又一次运行到 new_pro ，但是在 multiprocessing.Process 的源码中是对子进程再次产生子进程是做了限制的，是不允许的，于是便会出现错误。

2.2 继承创建

即继承Process来自定义进程类，重写run方法。

import multiprocessingfrom multiprocessing import Processimport time# 继承进程类class MyProcess(Process):def __init__(self, num):super().__init__() # 必须调用父类的初始化方法self.num = numdef run(self) -> None:while True:self.num += 2print('{} is running >> {}'.format(multiprocessing.current_process().name, self.num))time.sleep(2)if __name__ == '__main__':new_pro = MyProcess(100)# 设置进程名字new_pro.name = "Add2"new_pro.start()n = 0while True:n += 1print('{} is running >> {}'.format(multiprocessing.current_process().name, n))time.sleep(1)

3. 守护进程和join()方法

如果当前python进程是守护进程，那么意味着这个进程是“不重要”的，“不重要”意味着如果他的主进程结束了但该守护进程没有运行完，守护进程就会被强制结束。如果进程是非守护进程，那么父进程只有等到非守护进程运行完毕后才能结束。如果需要设置一个进程为守护进程，只需要将其 daemon 参数设置为 True 即可。

join 方法的参数如下：

join([timeout])

如果可选参数 timeout 是 None （默认值），则该方法将阻塞其他所有进程（包括主进程），直到调用 join() 方法的进程终止。如果 timeout 是一个正数，它最多会阻塞 timeout 秒。

import multiprocessingfrom multiprocessing import Processimport timedef process1(num):while True:num += 1print('{} is running >> {}'.format(multiprocessing.current_process().name, num))time.sleep(1)def process2(num):while True:num += 2print('{} is running >> {}'.format(multiprocessing.current_process().name, num))time.sleep(2)if __name__ == '__main__':new_pro1 = Process(target=process1, name='Add1', args=(100,))new_pro1.daemon = Truenew_pro1.start()new_pro1.join(3)new_pro2 = Process(target=process2, name='Add2', args=(1,))new_pro2.daemon = Truenew_pro2.start()time.sleep(3)print('{} Already Endding'.format(multiprocessing.current_process().name))

输出结果如下：

Add1 is running >> 101
Add1 is running >> 102
Add1 is running >> 103
Add2 is running >> 3
Add1 is running >> 104
Add1 is running >> 105
Add2 is running >> 5
Add1 is running >> 106
MainProcess Already Endding

以上结果显示，前三秒钟，Add1 阻塞了所有的进程，包括主进程，过了三秒后，阻塞结束，再过三秒后，主进程运行结束，由于没有了非守护进程，两个守护进程没有了守护的意义，故程序结束。

4. 进程锁

进程锁与线程锁不能说相似吧，也就是一模一样了 ，只不过是库变了一下，甚至连API都一样，仅仅是使用时由 lock = threading.Lock() 变为了 lock = multiprocessing.Lock() 等，具体的使用可以看多线程的这篇文章中的线程锁。

5. 进程通信

进程之间不共享数据的。如果进程之间需要进行通信，则要用到 Queue 模块或者 Pipe 模块来实现。

例如，我有两个进程，一个进程会产生一个随机数，另一个进程需要将上一个进程产生的随机数进行加1操作，我想大部分人都是想的是用全局变量来进行操作，操作如下：

from multiprocessing import Processimport randomdata = 0def process1():global datadata = random.random()print('产生的data = {}'.format(data))def process2():global datadata = data + 1print('加一后的data = {}'.format(data))if __name__ == '__main__':new_pro1 = Process(target=process1)new_pro1.start()new_pro2 = Process(target=process2)new_pro2.start()

输出结果为：

加一后的data = 1
产生的data = 0.885936521669484

很明显没有达到想要的目的。那么，为什么会这样呢？那是因为每个子进程享有独立的内存空间，接收进程产生的数据不能马上同步到转发进程中。所以，我们接下来讲讲使用 Queue 模块或者 Pipe 模块进行进程间的通信。

5.1 Queue

Queue 模块中最常用的方法就是 put 以及 get 方法了，一个是将数据放入队列中，一个是将数据从队列中取出。其他诸如 empty(),full() 等方法由于多线程或多进程的环境，这些方法是不可靠的，就不进行介绍了。

上述方法如下：

• Queue

  multiprocessing.Queue([maxsize])

  maxsize 参数可选，如果填入了参数，则申请一个 maxsize 大小的队列。

• put

  put(obj[, block[, timeout]])

  将 obj 放入队列。如果可选参数 block 是 True (默认值) 而且 timeout 是 None (默认值), 将会阻塞当前进程，直到有空的缓冲槽。如果 timeout 是正数，将会在阻塞了最多 timeout 秒之后还是没有可用的缓冲槽时抛出 queue.Full 异常。反之 (block 是 False 时)，仅当有可用缓冲槽时才放入对象，否则抛出 queue.Full 异常 (在这种情形下 timeout 参数会被忽略)。

• get

  get([block[, timeout]])

  从队列中取出并返回对象。如果可选参数 block 是 True (默认值) 而且 timeout 是 None (默认值), 将会阻塞当前进程，直到队列中出现可用的对象。如果 timeout 是正数，将会在阻塞了最多 timeout 秒之后还是没有可用的对象时抛出 queue.Empty 异常。反之 (block 是 False 时)，仅当有可用对象能够取出时返回，否则抛出 queue.Empty 异常 (在这种情形下 timeout 参数会被忽略)。

使用该模块将上面的例子写为进程间的通信后改为：

from multiprocessing import Process, Queueimport randomdef process1(q):data = random.random()# 将数据放入队列中q.put(data)print('产生的data = {}'.format(data))def process2(q):# 从队列中得到数据data = q.get()data = data + 1print('加一后的data = {}'.format(data))if __name__ == '__main__':# 初始化队列queue = Queue()new_pro1 = Process(target=process1, args=(queue,))new_pro1.start()new_pro2 = Process(target=process2, args=(queue,))new_pro2.start()

输出结果为：

产生的data = 0.2752695453210734
加一后的data = 1.2752695453210734

5.2 Pipe

如果你创建了很多个子进程，那么其中任何一个子进程都可以对Queue进行存（put）和取（get）。但Pipe不一样，Pipe只提供两个端点，只允许两个子进程进行存（send）和取（recv）。也就是说，Pipe实现了两个子进程之间的通信。

将上面的例子使用 Pipe 进行改动后，程序如下：

from multiprocessing import Process, Pipeimport randomdef process1(conn_1):data = random.random()# 将数据放入管道的一端conn_1.send(data)print('产生的data = {}'.format(data))def process2(conn_2):# 从管道另一端得到数据data = conn_2.recv()data = data + 1print('加一后的data = {}'.format(data))if __name__ == '__main__':# 初始化管道的两端conn_1, conn_2 = Pipe()new_pro1 = Process(target=process1, args=(conn_1,))new_pro1.start()new_pro2 = Process(target=process2, args=(conn_2,))new_pro2.start()

输出结果如下：

产生的data = 0.07282616926609853
加一后的data = 1.0728261692660985

6. 进程数据共享

上面的进程通信中， Queue 要求要先进先出， Pipe 只能够实现两个进程的存取，如果要一个数据拿给所有进程使用且不要求先进先出，那该使用什么呢？这就需要用到进程间的数据共享了。

6.1 Value

Value 数据共享类最多能够共享一个值，该函数的参数如下：

Value(typecode_or_type, args, lock=True)

上述方法中，参数 typecode_or_type 定义 ctypes() 对象的类型，可以传 Type code 或 C Type，具体对照表见下文。args 传递给 typecode_or_type 构造函数的参数，lock 默认为True，创建一个互斥锁来限制对Value对象的访问，如果传入一个锁，如Lock或RLock的实例，将用于同步。如果传入False，Value的实例就不会被锁保护，它将不是进程安全的。

例如下面我们在一个进程中传入值并对其进行改动，另一个进程输出传入的值，代码如下：

from multiprocessing import Process, Valuedef process1(n):n.value = 1def process2(n):print('改变后的参数 = {}'.format(n.value))if __name__ == '__main__':# 初始化valuenum = Value('d', 0)new_pro1 = Process(target=process1, args=(num,))new_pro1.start()new_pro1.join()new_pro2 = Process(target=process2, args=(num,))new_pro2.start()

输出如下：

改变后的参数 = 1.0

6.2 Array

上述的 Value 类只能传递一个参数，但是 Array 可以传递很多的参数，该方法参数如下：

Array(typecode_or_type, size_or_initializer, **kwds[, lock])

typecode_or_type 参数与上面的相同，同样参照上表，size_or_initializer 如果它是一个整数，那么它确定数组的长度，并且数组将被初始化为0。否则，size_or_initializer 是用于初始化数组的序列，其长度决定数组的长度。kwds 是传递给 typecode_or_type 构造函数的参数，lock 参数与上面的相同。

例如，我们传入一个0数组给一个进程，然后在另一个进程中计算其和，代码如下：

from multiprocessing import Process, Arraydef process1(n):n[3] = 0n[4] = 0def process2(n):# 从队列中得到数据print('数组的和为 = {}'.format(sum(n)))if __name__ == '__main__':# 初始化数组num = Array('d', range(5))new_pro1 = Process(target=process1, args=(num,))new_pro1.start()new_pro1.join()new_pro2 = Process(target=process2, args=(num,))new_pro2.start()

输出如下：

数组的和为 = 3.0

7. 进程池

进程池即可以提供指定数量的进程给用户使用,即当有新的请求提交到进程池中时，如果池未满，则会创建一个新的进程用来执行该请求;反之，如果池中的进程数已经达到规定最大值，那么该请求就会等待，只要池中有进程空闲下来，该请求就能得到执行。

进程池的话Python中有两个方法可以实现，一个是 multiprocessing 模块自带的 Pool 类，还有一个是 concurrent.futures.ProcessPoolExecutor 进行实现，其中后者的API与线程池的API一模一样，可以参考这里的线程池来进行学习，这里我介绍下前面一种方式。

该模块的常用方法如下：

• Pool

  Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])

  该类一般初始化使用的是 processes ，该参数是要使用的工作进程数目。如果 processes 为 None，则使用 os.cpu_count() 返回的值。

• apply

  apply(func[, args[, kwds]])

  使用阻塞的方式调用 func，必须等待上⼀个进程执行完任务后才能执行下一个进程，了解即可，几乎不用

• apply_async

  apply_async(func[, args[, kwds[, callback[, error_callback]]]])

  使用非阻塞的方式调用 func（任务并行执行)，args 为传递给 func 的参数列表，kwds 为传递给 func 的关键字参数列表。

• terminate

  terminate()

  不管任务是否完成，立即终止

• close

  close()

  关闭Pool，使其不再接受新的任务。

• join

  join()

  主进程阻塞，等待子进程的退出，必须在 close 或 terminate 之后使用。

下面创建一个容量为4的进程池，并让10个进程都停留三秒输出进程名，代码如下：

import multiprocessingfrom multiprocessing import Process, Poolimport timedef proc():time.sleep(3)print('{} Already Endding >> {}'.format(multiprocessing.current_process().name,time.strftime('%H:%M:%S',time.localtime(time.time()))))if __name__ == '__main__':# 创建容量为4的进程池pool = Pool(4)for i in range(10):pool.apply_async(proc)pool.close()# 阻塞主进程，等所有子进程运行完后再通过pool.join()print('{} Already Endding >> {}'.format(multiprocessing.current_process().name,time.strftime('%H:%M:%S', time.localtime(time.time()))))

输出如下：

SpawnPoolWorker-1 Already Endding >> 17:16:15
SpawnPoolWorker-4 Already Endding >> 17:16:15
SpawnPoolWorker-3 Already Endding >> 17:16:15
SpawnPoolWorker-2 Already Endding >> 17:16:15
SpawnPoolWorker-1 Already Endding >> 17:16:18
SpawnPoolWorker-2 Already Endding >> 17:16:18
SpawnPoolWorker-4 Already Endding >> 17:16:18
SpawnPoolWorker-3 Already Endding >> 17:16:18
SpawnPoolWorker-1 Already Endding >> 17:16:21
SpawnPoolWorker-3 Already Endding >> 17:16:21
MainProcess Already Endding >> 17:16:21

参考文章

[1] : https://docs.python.org/zh-cn/3.7/library/multiprocessing.html
[2] : https://www.freesion.com/article/2458937949/
[3] : https://blog.csdn.net/zong596568821xp/article/details/99678390
[4] : https://zhuanlan.zhihu.com/p/493699150
[5] : https://zhuanlan.zhihu.com/p/477826233
[6] : https://blog.csdn.net/weixin_56319791/article/details/122270700
[7] : https://zhuanlan.zhihu.com/p/568073350
[8] : https://www.jb51.net/article/230310.htm