Python中的并发编程-2

在上一课中我们提到，由于GIL（全局解释器锁）的存在，CPython中的多线程无法发挥CPU的多核优势。若想突破GIL的限制，可考虑使用多进程——对于多进程程序而言，每个进程都拥有独立的GIL，因此不会受到GIL的约束。那么，如何在Python程序中创建和使用多进程呢？

创建进程

在Python中，可基于Process类创建进程。尽管进程和线程存在本质区别，但Process类与之前讲解的Thread类用法非常相似。使用Process类构造器创建对象时，同样通过target参数传入函数，指定进程要执行的代码；通过args和kwargs参数，为目标函数传递位置参数和关键字参数。

from multiprocessing import Process, current_processfrom time import sleepdefsub_task(content, nums):# 通过current_process函数获取当前进程对象# 通过进程对象的pid和name属性获取进程的ID号和名字print(f'PID: {current_process().pid}')print(f'Name: {current_process().name}')# 通过下面的输出不难发现，每个进程都有自己的nums列表，进程之间本就不共享内存# 在创建子进程时复制了父进程的数据结构，三个进程从列表中pop(0)得到的值都是20    counter, total = 0, nums.pop(0)print(f'Loop count: {total}')    sleep(0.5)while counter < total:        counter += 1print(f'{counter}: {content}')        sleep(0.01)defmain():    nums = [20, 30, 40]# 创建并启动进程来执行指定的函数    Process(target=sub_task, args=('Ping', nums)).start()    Process(target=sub_task, args=('Pong', nums)).start()# 在主进程中执行sub_task函数    sub_task('Good', nums)if __name__ == '__main__':    main()

说明：上述代码通过current_process函数获取当前进程对象，再通过进程对象的pid属性获取进程ID。在Python中，使用os模块的getpid函数，也能达到同样的效果。

若有需求，也可使用os模块的fork函数创建进程。调用该函数时，操作系统会自动复制当前进程（父进程），生成一个新进程（子进程）：父进程中fork函数返回子进程的ID，子进程中fork函数返回0——也就是说，该函数调用一次，会在父进程和子进程中得到两个不同的返回值。需要注意的是，Windows系统不支持fork函数，若使用Linux或macOS系统，可尝试以下代码：

import osprint(f'PID: {os.getpid()}')pid = os.fork()if pid == 0:print(f'子进程 - PID: {os.getpid()}')print('Todo: 在子进程中执行的代码')else:print(f'父进程 - PID: {os.getpid()}')print('Todo: 在父进程中执行的代码')

简而言之，我们更推荐大家通过三种方式创建和使用多进程：直接使用Process类、继承Process类、使用进程池（ProcessPoolExecutor）。这三种方式不同于fork函数，能保证代码的兼容性和可移植性，具体做法与之前讲解的多线程创建方式较为接近，此处不再赘述。

多进程和多线程的比较

对于爬虫这类I/O密集型任务，使用多进程并无明显优势；但对于计算密集型任务，多进程相比多线程，效率会有显著提升，我们可通过以下代码验证这一点。下面的代码将通过多线程和多进程两种方式，判断一组大整数是否为质数——这是典型的计算密集型任务，我们将任务分配到多个线程和多个进程中，对比两者的执行表现。

首先实现多线程版本，代码如下：

import concurrent.futuresPRIMES = [1116281,1297337,104395303,472882027,533000389,817504243,982451653,112272535095293,112582705942171,112272535095293,115280095190773,115797848077099,1099726899285419] * 5defis_prime(n):"""判断素数"""for i inrange(2, int(n ** 0.5) + 1):if n % i == 0:returnFalsereturn n != 1defmain():"""主函数"""with concurrent.futures.ThreadPoolExecutor(max_workers=16) as executor:for number, prime inzip(PRIMES, executor.map(is_prime, PRIMES)):print('%d is prime: %s' % (number, prime))if __name__ == '__main__':    main()

假设上述代码保存为example.py文件，在Linux或macOS系统上，可使用time python example.py命令执行程序，同时获取操作系统的执行时间统计。在我的macOS设备上，某次运行结果的最后一行如下：

python example09.py  38.69s user 1.01s system 101% cpu 39.213 total

从结果可以看出，多线程代码的CPU利用率仅能达到100%，这进一步证明了多线程无法利用CPU多核特性加速执行。接下来我们看多进程版本，只需将上述代码中的线程池（ThreadPoolExecutor）替换为进程池（ProcessPoolExecutor）即可。

多进程的版本

import concurrent.futuresPRIMES = [1116281,1297337,104395303,472882027,533000389,817504243,982451653,112272535095293,112582705942171,112272535095293,115280095190773,115797848077099,1099726899285419] * 5defis_prime(n):"""判断素数"""for i inrange(2, int(n ** 0.5) + 1):if n % i == 0:returnFalsereturn n != 1defmain():"""主函数"""with concurrent.futures.ProcessPoolExecutor(max_workers=16) as executor:for number, prime inzip(PRIMES, executor.map(is_prime, PRIMES)):print('%d is prime: %s' % (number, prime))if __name__ == '__main__':    main()

提示：运行上述代码时，可通过操作系统的任务管理器（或资源监视器），查看是否启动了多个Python解释器进程。

我们仍使用time python example.py命令执行该代码，运行结果的最后一行如下：

python example09.py 106.63s user 0.57s system 389% cpu 27.497 total

可以看出，在我的设备上，多进程版本的CPU利用率接近400%；用户态CPU耗时（106.63秒）几乎是程序总运行时间（27.497秒）的4倍——这两点均说明我的设备搭载了4核CPU。当然，若想查看自己设备的CPU核心数，可直接运行以下代码：

import osprint(os.cpu_count())

综上所述，多进程可突破GIL限制，充分利用CPU多核特性，这对于计算密集型任务至关重要。常见的计算密集型任务包括科学计算、图像处理、音视频编解码等，若这类任务本身支持并行执行，使用多进程会是更优选择。

进程间通信

在讲解进程间通信之前，先给大家一个任务：启动两个进程，一个输出“Ping”，一个输出“Pong”，当两个进程输出的“Ping”和“Pong”总数达到50个时，终止程序。看似简单，但实际编写代码时会发现，由于多进程无法像多线程那样通过共享内存交换数据，以下代码无法达到预期效果：

from multiprocessing import Processfrom time import sleepcounter = 0defsub_task(string):global counterwhile counter < 50:print(string, end='', flush=True)        counter += 1        sleep(0.01)defmain():    Process(target=sub_task, args=('Ping', )).start()    Process(target=sub_task, args=('Pong', )).start()if __name__ == '__main__':    main()

上述代码看似无误，但最终结果是“Ping”和“Pong”各输出50个。再次提醒大家：创建子进程时，子进程会复制父进程的所有数据结构，每个子进程拥有独立的内存空间——这意味着两个子进程各有一个counter变量，且都会从0加到50，因此出现上述结果。

解决该问题的简单方法，是使用multiprocessing模块中的Queue类。它是可被多进程共享的队列，底层通过操作系统的管道和信号量（semaphore）机制实现，具体代码如下：

import timefrom multiprocessing import Process, Queuedefsub_task(content, queue):    counter = queue.get()while counter < 50:print(content, end='', flush=True)        counter += 1        queue.put(counter)        time.sleep(0.01)        counter = queue.get()defmain():    queue = Queue()    queue.put(0)    p1 = Process(target=sub_task, args=('Ping', queue))    p1.start()    p2 = Process(target=sub_task, args=('Pong', queue))    p2.start()while p1.is_alive() and p2.is_alive():pass    queue.put(50)if __name__ == '__main__':    main()

提示：multiprocessing.Queue对象的get方法，默认在队列为空时会阻塞，直到获取到数据才返回。若不希望阻塞，或需要指定阻塞超时时间，可通过block和timeout参数设置。

上述代码通过Queue类的get和put方法，实现了三个进程（p1、p2和主进程）的数据共享，这就是进程间通信。当从Queue中取出的值大于等于50时，p1和p2会跳出while循环，终止进程执行。代码第22行的循环用于等待p1和p2中的一个进程结束，此时主进程需向Queue中放入一个大于等于50的值，确保另一个未结束的进程读取到该值后也能正常终止。

进程间通信的方式还有很多，例如使用套接字也可实现多进程通信，甚至支持不同主机上的进程通信，有兴趣的读者可自行深入了解。

总结

在Python中，还可通过subprocess模块的call函数，执行其他命令创建子进程——相当于在当前程序中调用其他程序，此处暂不深入探讨，有兴趣的读者可自行研究。

对于Python开发者而言，以下情况可考虑使用多线程：

1. 1. 程序需要维护大量共享状态（尤其是可变状态）。Python中的列表、字典、集合均为线程安全的（多个线程同时操作同一个列表、字典或集合，不会引发错误和数据异常），因此使用线程维护共享状态的代价，相对低于使用进程。
2. 2. 程序大部分时间用于I/O操作，无过多并行计算需求，且不占用大量内存。

而遇到下列情况时，应考虑使用多进程：

1. 1. 程序执行计算密集型任务（如音视频编解码、数据压缩、科学计算等）。
2. 2. 程序的输入可并行拆分，且运算结果可合并。
3. 3. 程序在内存使用上无限制，且不强烈依赖I/O操作（如读写文件、套接字等）。