Python多进程的应用场景和使用方法

并行和串行计算

想象一下，你有一个巨大的问题要解决，而你独自一人。你需要计算八个不同数字的平方根。你是做什么的？你没有太多选择。从第一个数字开始，然后计算结果。然后，你继续和其他人。

如果你有三个擅长数学的朋友愿意帮助你呢？他们每个人都会计算两个数字的平方根，你的工作会更容易，因为工作量在你的朋友之间平均分配。这意味着你的问题将更快地得到解决。

好了，一切都清楚了吗？在这些示例中，每个朋友代表CPU的核心。在第一个示例中，整个任务由你依次解决。这称为串行计算。在第二个示例中，由于你总共使用了四个内核，因此你使用的是并行计算。并行计算涉及使用并行进程或在处理器的多个核之间划分的进程。

并行编程模型

我们已经确定了什么是并行编程，但我们如何使用它？我们之前说过，并行计算涉及在处理器的多个核心之间执行多个任务，这意味着这些任务是同时执行的。在进行并行化之前，你应该考虑几个问题。例如，是否有其他优化可以加快我们的计算速度？

现在，让我们理所当然地认为并行化是最适合的解决方案。并行计算主要有三种模式：

• 完全平行。任务可以独立运行，不需要相互通信。

• 共享内存并行。进程（或线程）需要通信，因此它们共享一个全局地址空间。

• 消息传递。进程需要在需要时共享消息。

在本文中，我们将说明第一个模型，它也是最简单的。

Python多进程：Python中基于进程的并行性

在 Python 中实现并行性的一种方法是使用multiprocessing 模块。multiprocessing模块允许你创建多个进程，每个进程都有自己的 Python 解释器。因此，Python 多进程实现了基于进程的并行。

你可能听说过其他库，比如threading，它也是Python内置的，但它们之间有着重要的区别。multiprocessing模块创建新进程，而threading创建新线程。

使用多进程的好处

你可能会问，“为什么选择多进程？” 多进程可以通过并行而不是按顺序运行多个任务来显着提高程序的效率。一个类似的术语是多线程，但它们是不同的。

进程是加载到内存中运行的程序，不与其他进程共享其内存。线程是进程中的一个执行单元。多个线程在一个进程中运行，并相互共享进程的内存空间。

Python的全局解释器锁（GIL）只允许在解释器下一次运行一个线程，这意味着如果需要Python解释器，你将无法享受多线程的性能优势。这就是在Python中多进程比线程更占优势的原因。多个进程可以并行运行，因为每个进程都有自己的解释器，执行分配给它的指令。此外，操作系统将在多个进程中查看你的程序，并分别对它们进行调度，即，你的程序在总的计算机资源中占有更大的份额。因此，当程序受到CPU限制时，多进程速度更快。在程序中有大量I/O的情况下，线程可能更高效，因为大多数时候，程序都在等待I/O完成。然而，多进程通常效率更高，因为它同时运行。

以下是多进程的一些好处：

• 在处理高CPU密集型任务时更好地使用CPU

• 与线程相比，对子线程的控制更多

• 易于编码

第一个优点与性能有关。由于多进程创建了新的进程，你可以通过在其他内核之间划分任务来更好地利用CPU的计算能力。现在大多数处理器都是多核处理器，如果你优化代码，可以通过并行计算节省时间。

第二个优点是多线程处理的替代方案。线程不是进程，这有其后果。如果你创建了一个线程，那么像处理正常进程一样终止它甚至中断它是很危险的。由于多进程和多线程之间的比较不在本文的范围内，后续我会单独写一篇来讲讲多进程和多线程的区别。

多进程的第三个优点是它很容易实现，因为你尝试处理的任务适合并行编程。

Python多进程入门

我们终于准备好编写一些 Python 代码了！

我们将从一个非常基本的示例开始，我们将使用它来说明 Python 多进程的核心方面。在此示例中，我们将有两个进程：

• parent经常。只有一个父进程，它可以有多个子进程。

• child进程。这是由父进程产生的。每个子进程也可以有新的子进程。

我们将使用该child过程来执行某个函数。这样，parent可以继续执行。

一个简单的 Python多进程示例

这是我们将用于此示例的代码：

from multiprocessing import Process

def bubble_sort(array):
    check = True
    while check == True:
      check = False
      for i in range(0, len(array)-1):
        if array[i] > array[i+1]:
          check = True
          temp = array[i]
          array[i] = array[i+1]
          array[i+1] = temp
    print("Array sorted: ", array)

if __name__ == '__main__':
    p = Process(target=bubble_sort, args=([1,9,4,5,2,6,8,4],))
    p.start()
    p.join()

在这个片段中，我们定义了一个名为bubble_sort(array)。这个函数是冒泡排序算法的一个非常简单的实现。如果你不知道它是什么，请不要担心，因为它并不重要。要知道的关键是它是一个可以实现某个功能的函数。

进程类

从multiprocessing，我们导入类Process。此类表示将在单独进程中运行的活动。事实上，你可以看到我们已经传递了一些参数：

• target=bubble_sort，意味着我们的新进程将运行该bubble_sort函数

• args=([1,9,4,52,6,8,4],)，这是作为参数传递给目标函数的数组

一旦我们创建了 Process 类的实例，我们只需要启动该进程。这是通过编写p.start()完成的。此时，该进程开始。

在我们退出之前，我们需要等待子进程完成它的计算。该join()方法等待进程终止。

在这个例子中，我们只创建了一个子进程。正如你可能猜到的，我们可以通过在Process类中创建更多实例来创建更多子进程。

进程池类

如果我们需要创建多个进程来处理更多 CPU 密集型任务怎么办？我们是否总是需要明确地开始并等待终止？这里的解决方案是使用Pool类。

Pool类允许你创建一个工作进程池，在下面的示例中，我们将研究如何使用它。这是我们的新示例：

from multiprocessing import Pool
import time
import math

N = 5000000

def cube(x):
    return math.sqrt(x)

if __name__ == "__main__":
    with Pool() as pool:
      result = pool.map(cube, range(10,N))
    print("Program finished!")

在这个代码片段中，我们有一个cube(x)函数，它只接受一个整数并返回它的平方根。很简单，对吧？

然后，我们创建一个Pool类的实例，而不指定任何属性。默认情况下，Pool类为每个 CPU 核心创建一个进程。接下来，我们使用几个参数运行map方法。

map方法将cube函数应用于我们提供的可迭代对象的每个元素——在本例中，它是从10到N的每个数字的列表。

这样做的最大优点是列表上的计算是并行进行的！

joblib

包joblib是一组使并行计算更容易的工具。它是一个用于多进程的通用第三方库。它还提供缓存和序列化功能。要安装joblib包，请在终端中使用以下命令：

pip install joblib

我们可以将之前的示例转换为以下示例以供使用joblib：

from joblib import Parallel, delayed
 
def cube(x):
    return x**3
 
start_time = time.perf_counter()
result = Parallel(n_jobs=3)(delayed(cube)(i) for i in range(1,1000))
finish_time = time.perf_counter()
print(f"Program finished in {finish_time-start_time} seconds")
print(result)

事实上，直观地看到它的作用。delayed()函数是另一个函数的包装器，用于生成函数调用的“延迟”版本。这意味着它在被调用时不会立即执行函数。

然后，我们多次调用delayed函数，并传递不同的参数集。例如，当我们将整数1赋予cube函数的延迟版本时，我们不计算结果，而是分别为函数对象、位置参数和关键字参数生成元组(cube, (1,), {})。

我们使用Parallel()创建了引擎实例。当它像一个以元组列表作为参数的函数一样被调用时，它将实际并行执行每个元组指定的作业，并在所有作业完成后收集结果作为列表。在这里，我们创建了n_jobs=3的Parallel()实例，因此将有三个进程并行运行。

我们也可以直接编写元组。因此，上面的代码可以重写为：

result = Parallel(n_jobs=3)((cube, (i,), {}) for i in range(1,1000))

使用joblib的好处是，我们可以通过简单地添加一个附加参数在多线程中运行代码：

result = Parallel(n_jobs=3, prefer="threads")(delayed(cube)(i) for i in range(1,1000))

这隐藏了并行运行函数的所有细节。我们只是使用与普通列表理解没有太大区别的语法。

充分利用 Python多进程

创建多个进程并进行并行计算不一定比串行计算更有效。对于 CPU 密集度较低的任务，串行计算比并行计算快。因此，了解何时应该使用多进程非常重要——这取决于你正在执行的任务。

为了让你相信这一点，让我们看一个简单的例子：

from multiprocessing import Pool
import time
import math

N = 5000000

def cube(x):
    return math.sqrt(x)

if __name__ == "__main__":
    # first way, using multiprocessing
    start_time = time.perf_counter()
    with Pool() as pool:
      result = pool.map(cube, range(10,N))
    finish_time = time.perf_counter()
    print("Program finished in {} seconds - using multiprocessing".format(finish_time-start_time))
    print("---")
    # second way, serial computation
    start_time = time.perf_counter()
    result = []
    for x in range(10,N):
      result.append(cube(x))
    finish_time = time.perf_counter()
    print("Program finished in {} seconds".format(finish_time-start_time))

此代码段基于前面的示例。我们正在解决同样的问题，即计算N个数的平方根，但有两种方法。第一个涉及 Python 进程的使用，而第二个不涉及。我们使用time库中的perf_counter()方法来测量时间性能。

在我的电脑上，我得到了这个结果：

> python code.py
Program finished in 1.6385094 seconds - using multiprocessing
---
Program finished in 2.7373942999999996 seconds

如你所见，相差不止一秒。所以在这种情况下，多进程更好。

让我们更改代码中的某些内容，例如N的值。 让我们把它降低到N=10000，看看会发生什么。

这就是我现在得到的：

> python code.py
Program finished in 0.3756742 seconds - using multiprocessing
---
Program finished in 0.005098400000000003 seconds

发生了什么？现在看来，多进程是一个糟糕的选择。为什么？

与解决的任务相比，在进程之间拆分计算所带来的开销太大了。你可以看到在时间性能方面有多大差异。