Python 高手编程系列三千零三：多进程

老实说，多线程是很有挑战性的-我们已经在上一节中看到了。事实上，对问题的最简
单的方法是只需要最小的代价。但是以一种安全的方式处理线程需要大量的代码。
我们必须设置线程池和通信队列，优雅地处理来自线程的异常，并且在尝试提供速率
限制功能时也考虑线程安全。十行代码只能从外部库并行执行一个函数！我们假设它可以
用于生产环境，因为有外部包创建者的承诺，它的库是线程安全的。听起来像一个高价格
的解决方案，实际上它只适用于执行 I/O 绑定任务。
实现并行性的另一种方法是多进程。彼此独立 Python 进程没有 GIL 的限制，这样可以
有更好的资源利用率。这对于在多核处理器上运行的应用程序尤其重要，这些处理器可以
真正的处理 CPU 密集型任务。现在这是为 Python 开发人员提供的唯一内置并行解决方案
（使用 CPython 解释器），你可以从多个处理器核心中受益。
使用多个进程的另一个优点是它们不共享内存上下文。因此，很难破坏数据也难以在
应用程序中引入死锁。不共享内存上下文意味着你需要一些额外的努力在隔离的进程之间
传递数据，但幸运的是有许多好的方法来实现可靠的进程间通信。事实上，Python 提供了
一些原语，使进程之间的通信与线程之间的一样简单。
在任何编程语言中启动新进程的最基本的方法通常是在某个时刻派生程序。在 POSIX
系统（Unix、Mac OS 和 Linux）上，派生是通过 os.fork()函数在 Python 中暴露的系统
调用，它将创建一个新的子进程。然后两个进程在派生后自己继续该程序。以下是一个示
例脚本，它自己派生一次：
import os
pid_list = []
def main():
pid_list.append(os.getpid())
child_pid = os.fork()
if child_pid == 0:
pid_list.append(os.getpid())
print()
print(“CHLD: hey, I am the child process”)
print(“CHLD: all the pids i know %s” % pid_list)
else:
pid_list.append(os.getpid())
print()
print(“PRNT: hey, I am the parent”)
print(“PRNT: the child is pid %d” % child_pid)
print(“PRNT: all the pids i know %s” % pid_list)
ifname== “main”:
main()
以下是一个在终端中运行它的例子：
$ python3 forks.py
PRNT: hey, I am the parent
PRNT: the child is pid 21916
PRNT: all the pids i know [21915, 21915]
CHLD: hey, I am the child process
CHLD: all the pids i know [21915, 21916]
注意这两个进程在 os.fork()调用之前它们的数据具有完全相同的初始状态。它们都具
有与 pid_list 集合的第一个值相同的 PID 号（进程标识符）。后来，两个状态发生了分歧，
我们可以看到子进程添加了 21916 值，而父进程复制了它的 21915 PID。这是因为这两个进
程的内存上下文不共享。它们具有相同的初始条件，但在 os.fork()调用后不能相互影响。
派生将内存上下文复制到子进程后，每个进程都会处理自己的地址空间。为了沟通，
进程需要与系统范围的资源或使用低级工具（如信号）。
不幸的是，os.fork 在 Windows 下不可用，需要生成一个新的解释器以模仿 fork
功能。所以它根据不同的平台会有差别。os 模块还暴露了函数，它可以在 Windows 下生
成新进程，但最终你很少使用它们。os.fork()也是如此。Python 提供了一个很好的
multiprocessing 模块，为多进程创建了一个高级接口。这个模块的最大优点是它提供
了一些抽象，这些抽象针对我们必须从头开始编写一个多线程应用的例子。它可以限制样
板代码的数量，从而提高应用程序可维护性并降低其复杂性。令人惊讶的是，尽管它的名
称，multiprocessing 模块也暴露了类似的线程接口，所以你可能想要使用相同的接口
来实现两种方法。
内置的 multiprocessing 模块
multiprocessing 提供了一种便捷的方式来处理进程，就像它们是线程一样。
此模块包含一个与 Thread 类非常相似的 Process 类，可以在任何平台上使用：
from multiprocessing import Process
import os
def work(identifier):
print(
‘hey, i am a process {}, pid: {}’
‘’.format(identifier, os.getpid())
)
def main():
processes = [
Process(target=work, args=(number,))
for number in range(5)
]
for process in processes:
process.start()
while processes:
processes.pop().join()
ifname== “main”:
main()
上述脚本在执行时会输出以下结果：
$ python3 processing.py
hey, i am a process 1, pid: 9196
hey, i am a process 0, pid: 8356
hey, i am a process 3, pid: 9524
hey, i am a process 2, pid: 3456
hey, i am a process 4, pid: 6576
当创建进程时，内存被派生（在 POSIX 系统上）。最有效的进程用法是让它们在创建
后自己工作以避免开销，并从主线程检查它们的状态。除了被复制的内存状态之外，
Process 类还在其构造函数中提供了一个额外的 args 参数，以便传递数据。
进程模块之间的通信需要一些额外的工作，因为它们的本地内存在默认情况下不共享。
为了简化这一点，multiprocessing 模块提供了进程之间的几种通信方式：
• 使用 multiprocessing.Queue 类，它是早先用于线程之间通信的 queue.Queue
的近似克隆。
• 使用 multiprocessing.Pipe，这是一个类似于套接字的双向通信通道。
• 使用 multiprocessing.sharedctypes 模块，通过它可以在进程之间共享的
专用内存池中创建任意 C 类型（从 ctypes 模块）。
multiprocessing.Queue 和 queue.Queue 类具有相同的接口。唯一的区别是第
一个是设计用于多进程环境，而不是多个线程，所以它使用不同的内部传输和锁定原语。
我们已经在一个多线程应用的例子中了解了如何在多线程中使用 Queue，因此我们不会用
多进程执行相同的操作。使用保持完全相同，所以这样的例子不会带来任何新的内容。
现在一个更有趣的模式是由 Pipe 类提供的。它是一个双工（双向）通信通道，在概念上非常类似于 Unix 管道。管道的接口也非常类似于来自内置 socket 模块的简单套接字。
与原始系统管道和套接字的区别在于，你可以发送任何可选对象（使用 pickle 模块），而
不仅是原始字节。这使得进程之间可以更容易的通信，因为你几乎可以发送任何基本的
Python 类型，如下所示：
from multiprocessing import Process, Pipe
class CustomClass:
pass
def work(connection):
while True:
instance = connection.recv()
if instance:
print(“CHLD: {}”.format(instance))
else:
return
def main():
parent_conn, child_conn = Pipe()
child = Process(target=work, args=(child_conn,))
for item in (
42,
‘some string’,
{‘one’: 1},
CustomClass(),
None,
):
print(“PRNT: send {}:”.format(item))
parent_conn.send(item)
child.start()
child.join()
ifname== “main”:
main()
当查看上述脚本的示例输出时，你将看到，你可以轻松地传递自定义类实例，并且它们具有不同的地址，具体取决于进程如下所示：
PRNT: send: 42
PRNT: send: some string
PRNT: send: {‘one’: 1}
PRNT: send: <main.CustomClass object at 0x101cb5b00>
PRNT: send: None
CHLD: recv: 42
CHLD: recv: some string
CHLD: recv: {‘one’: 1}
CHLD: recv: <main.CustomClass object at 0x101cba400>
另一种在进程之间共享状态的方法是在 multiprocessing.sharedctypes 中提供的
类中使用共享内存池中的原始类型。最基本的是 Value 和 Array。下面是 multiprocessing
模块的官方文档中的示例代码：
from multiprocessing import Process, Value, Array
def f(n, a):
n.value = 3.1415927
for i in range(len(a)):
a[i] = -a[i]
ifname== ‘main’:
num = Value(‘d’, 0.0)
arr = Array(‘i’, range(10))
p = Process(target=f, args=(num, arr))
p.start()
p.join()
print(num.value)
print(arr[:])
此示例将打印以下输出：
3.1415927
[0, -1, -2, -3, -4, -5, -6, -7, -8, -9]
使用 multiprocessing.sharedctypes 时，你需要记住，你正在处理共享内存，
或其他进程间通信通道。在大多数情况下，避免共享类型是合理的，因为它们增加代码复
杂性并带来多线程中已知的所有危险。