Python 并发编程 - 多进程

---

---

本章我们将重点比较多进程和多线程。

多进程

它是在单个计算机系统中使用两个或多个 CPU 单元。这是充分发挥硬件潜力的最佳方法，可以利用计算机系统中可用的所有 CPU 核心。

多线程

这是 CPU 管理操作系统使用情况的能力，通过并发执行多个线程来实现。多线程的主要思想是通过将进程划分为多个线程来实现并行。

下表显示了它们之间的一些重要区别：

消除全局解释器锁 (GIL) 的影响

在处理并发应用程序时，Python 中存在一个名为GIL（全局解释器锁）的限制。GIL 从不允许我们利用 CPU 的多个核心，因此可以说 Python 中没有真正的线程。GIL 是互斥锁，它使事物线程安全。换句话说，我们可以说 GIL 阻止多个线程并行执行 Python 代码。一次只能由一个线程持有锁，如果要执行线程，则必须首先获取锁。

使用多进程，我们可以有效地绕过 GIL 造成的限制：

• 通过使用多进程，我们正在利用多个进程的能力，因此我们正在利用 GIL 的多个实例。

• 因此，在任何时候都没有限制在我们的程序中执行一个线程的字节码。

在 Python 中启动进程

可以使用以下三种方法在 multiprocessing 模块中启动 Python 中的进程：

• fork
• spawn
• forkserver

使用 fork 创建进程

fork 命令是 UNIX 中的标准命令。它用于创建称为子进程的新进程。此子进程与称为父进程的进程并发运行。这些子进程与其父进程也相同，并继承父进程可用的所有资源。创建具有 fork 的进程时，使用以下系统调用：

• fork() - 它通常在内核中实现的系统调用。它用于创建进程的副本。

• getpid() - 此系统调用返回调用进程的进程 ID (PID)。

示例

下面的 Python 脚本示例将帮助您了解如何创建新的子进程并获取子进程和父进程的 PID：

import os

def child():
   n = os.fork()
   
   if n > 0:
      print("PID of Parent process is : ", os.getpid())

   else:
      print("PID of Child process is : ", os.getpid())
child()

输出

PID of Parent process is : 25989
PID of Child process is : 25990

使用 spawn 创建进程

spawn 意味着启动新事物。因此，生成进程意味着父进程创建新进程。父进程继续异步执行或等待子进程结束执行。按照以下步骤生成进程：

• 导入 multiprocessing 模块。

• 创建进程对象。

• 通过调用start()方法启动进程活动。

• 等待进程完成其工作并通过调用join()方法退出。

示例

下面的 Python 脚本示例有助于生成三个进程

import multiprocessing

def spawn_process(i):
   print ('This is process: %s' %i)
   return

if __name__ == '__main__':
   Process_jobs = []
   for i in range(3):
   p = multiprocessing.Process(target = spawn_process, args = (i,))
      Process_jobs.append(p)
   p.start()
   p.join()

输出

This is process: 0
This is process: 1
This is process: 2

使用 forkserver 创建进程

forkserver 机制仅适用于支持通过 Unix 管道传递文件描述符的某些 UNIX 平台。请考虑以下几点，以了解 forkserver 机制的运作方式：

• 使用 forkserver 机制启动新进程时，会实例化一个服务器。

• 然后，服务器接收命令并处理所有创建新进程的请求。

• 为了创建一个新进程，我们的 Python 程序将向 forkserver 发送请求，它将为我们创建一个进程。

• 最后，我们可以在程序中使用这个新创建的进程。

Python 中的守护进程

Python 的multiprocessing模块允许我们通过其 daemonic 选项拥有守护进程。在后台运行的守护进程或进程遵循与守护线程类似的概念。要执行后台进程，我们需要将 daemonic 标志设置为 true。守护进程将在主进程执行期间继续运行，并在完成执行或主程序被终止后终止。

示例

这里，我们使用与守护线程中使用的相同的示例。唯一的区别是将模块从multithreading更改为multiprocessing并将 daemonic 标志设置为 true。但是，输出将发生变化，如下所示：

import multiprocessing
import time

def nondaemonProcess():
   print("starting my Process")
   time.sleep(8)
   print("ending my Process")
def daemonProcess():
   while True:
   print("Hello")
   time.sleep(2)
if __name__ == '__main__':
   nondaemonProcess = multiprocessing.Process(target = nondaemonProcess)
   daemonProcess = multiprocessing.Process(target = daemonProcess)
   daemonProcess.daemon = True
   nondaemonProcess.daemon = False
   daemonProcess.start()
   nondaemonProcess.start()

输出

starting my Process
ending my Process

与守护线程生成的输出相比，输出不同，因为非守护模式下的进程没有输出。因此，守护进程在主程序结束后会自动结束，以避免持久运行进程。

在 Python 中终止进程

我们可以使用terminate()方法立即终止进程。我们将使用此方法在完成执行之前立即终止使用函数创建的子进程。

示例

import multiprocessing
import time
def Child_process():
   print ('Starting function')
   time.sleep(5)
   print ('Finished function')
P = multiprocessing.Process(target = Child_process)
P.start()
print("My Process has terminated, terminating main thread")
print("Terminating Child Process")
P.terminate()
print("Child Process successfully terminated")

输出

My Process has terminated, terminating main thread
Terminating Child Process
Child Process successfully terminated

输出显示程序在使用 Child_process() 函数创建的子进程执行之前终止。这意味着子进程已成功终止。

识别 Python 中的当前进程

操作系统中的每个进程都有称为 PID 的进程标识。在 Python 中，我们可以使用以下命令找到当前进程的 PID：

import multiprocessing
print(multiprocessing.current_process().pid)

示例

下面的 Python 脚本示例有助于找出主进程的 PID 以及子进程的 PID：

import multiprocessing
import time
def Child_process():
   print("PID of Child Process is: {}".format(multiprocessing.current_process().pid))
print("PID of Main process is: {}".format(multiprocessing.current_process().pid))
P = multiprocessing.Process(target=Child_process)
P.start()
P.join()

输出

PID of Main process is: 9401
PID of Child Process is: 9402

在子类中使用进程

我们可以通过子类化threading.Thread类来创建线程。此外，我们还可以通过子类化multiprocessing.Process类来创建进程。要在子类中使用进程，我们需要考虑以下几点：

• 我们需要定义Process类的新的子类。

• 我们需要重写_init_(self [,args] )类。

• 我们需要重写run(self [,args] )方法来实现Process的功能。

• 我们需要通过调用start()方法启动进程。

示例

import multiprocessing
class MyProcess(multiprocessing.Process):
   def run(self):
   print ('called run method in process: %s' %self.name)
   return
if __name__ == '__main__':
   jobs = []
   for i in range(5):
   P = MyProcess()
   jobs.append(P)
   P.start()
   P.join()

输出

called run method in process: MyProcess-1
called run method in process: MyProcess-2
called run method in process: MyProcess-3
called run method in process: MyProcess-4
called run method in process: MyProcess-5

Python 多进程模块 – Pool 类

如果我们在 Python 应用程序中讨论简单的并行处理任务，那么 multiprocessing 模块为我们提供 Pool 类。Pool类的以下方法可用于在主程序中启动多个子进程

apply() 方法

此方法类似于.ThreadPoolExecutor的.submit()方法。它会阻塞，直到结果准备就绪。

apply_async() 方法

当我们需要并行执行任务时，我们需要使用apply_async()方法将任务提交到池。这是一个异步操作，在所有子进程执行完毕之前不会锁定主线程。

map() 方法

与apply()方法一样，它也会阻塞，直到结果准备就绪。它等同于内置的map()函数，该函数将可迭代数据分成多个块，并将其作为单独的任务提交到进程池。

map_async() 方法

它是map()方法的一个变体，就像apply_async()对于apply()方法一样。它返回一个结果对象。当结果准备就绪时，会对其应用一个可调用对象。可调用对象必须立即完成；否则，处理结果的线程将被阻塞。

示例

下面的示例将帮助您实现一个用于执行并行执行的进程池。通过multiprocessing.Pool方法应用square()函数来执行简单的数字平方计算。然后使用pool.map()提交 5，因为输入是从 0 到 4 的整数列表。结果将存储在p_outputs中并打印出来。

def square(n):
   result = n*n
   return result
if __name__ == '__main__':
   inputs = list(range(5))
   p = multiprocessing.Pool(processes = 4)
   p_outputs = pool.map(function_square, inputs)
   p.close()
   p.join()
   print ('Pool :', p_outputs)

输出

Pool : [0, 1, 4, 9, 16]