前言
前面我们了解了计算机中的CPU多任务原理，知道了进程和线程的概念。今天就通过代码看Python中多进程是如何实现的。
Linux\Unix多进程：os.fork()
Linux\Unix操作系统提供了一个fork()系统调用，它非常特殊。普通的函数调用，调用一次，返回一次，但是fork()调用一次，返回两次，因为操作系统自动把当前进程（称为父进程）复制了一份（称为子进程），然后，分别在父进程和子进程内返回。
fork()返回结果：子进程永远返回0，而父进程返回子进程的ID。这样做的理由是，一个父进程可以fork出很多子进程，所以，父进程要记下每个子进程的ID，而子进程只需要调用getppid()就可以拿到父进程的ID。
Python的os模块封装了常见的系统调用，其中就包括fork，可以在Python程序中轻松创建子进程：
import os
pid = os.fork()
if pid == 0:
print("这个是子进程，pid是{}, 父进程pid是{}".format(os.getpid(), os.getppid()))
else:
print("这个是父进程，pid是{}, 创建了个子进程pid是{}".format(os.getpid(), pid))
提示：os.fork()只在Linux\Unix系统下可以使用。Mac系统是基于Unix系统开发的，所以也支持这种写法。如果你是在Windows系统下做的Python开发，那么运行上面代码，会报错：
Traceback (most recent call last):
File "F:/python_projects/multi_process/process_thread.py", line 2, in <module>
pid = os.fork()
AttributeError: module 'os' has no attribute 'fork'
由于Windows没有fork调用，但Python是支持跨平台的，所以Windows上用Python编写多进程的程序，就需要借用multiprocessing模块了。
跨平台多进程：multiprocessing
multiprocessing模块提供了一个Process类来代表一个进程对象。通常可导入模块：
from multiprocessing import Process
示例：
from multiprocessing import Process
from time import sleep
import os
# 定义函数
def my_run(arg):
print("子进程启动")
sleep(5)
print("子进程结束")
if __name__ == "__main__":
print("主进程已经开始启动:{}".format(os.getpid()))
# 创建子进程
p = Process(target=my_run, args=("子进程",))
# 启动子进程
p.start()
# join()方法可以等待子进程结束后再继续往下运行，通常用于进程间的同步
p.join()
print("父进程结束")
运行结果：
主进程已经开始启动:26308
子进程启动
子进程结束
父进程结束
提示：上面例子中join()方法可以等待子进程结束后再继续往下运行，通常用于进程间的同步。说白了就是，父进程的结束时间是要等子进程结束后才结束。如果不调用此方法，运行结果如下：
主进程已经开始启动:960
父进程结束
子进程启动
子进程结束
父进程的结束时间不再受子进程结束的影响。父进程可能在子进程结束之前就结束，也可能在子进程结束后才结束，具体情况要看各自进程的具体操作。本例中，如果不调用p.join()方法，但子进程有耗时操作sleep(5)，而父进程中没有耗时的操作，所以父进程优先于子进程结束了。
说明：
1，创建子进程时，只需要传入一个执行函数和函数的参数，创建一个Process实例，用start()方法启动，这样创建进程比fork()还要简单。
2.创建子进程时，切不可这样创建：p = Process(target=my_run("子进程"))。为了方便对比，将两种写法拿出来：
# 上面例子中的创建写法
p = Process(target=my_run, args=("子进程",))
# 不推荐的创建写法
p = Process(target=my_run("子进程"))
参数的传入方式不同，上面例子中通过args=传入，形如：args=("子进程",)，而错误的写法是直接通过指定target=时就一起传入，形如：target=my_run("子进程")。
错误的创建形式虽然可以创建子进程成功，但会导致：join()方法无论是否调用，都会出现进程同步的现象。即：父进程的结束时间是要等子进程结束后才结束。
线程池Pool
如果要启动大量的子进程，可以用进程池的方式批量创建子进程。multiprocessing模块提供了一个Pool类来代表一个进程对象。通常可导入模块：
from multiprocessing import Pool
Pool类可以提供指定数量的进程供用户调用，当有新的请求提交到Pool中时，如果池还没有满，就会创建一个新的进程来执行请求。如果池满，请求就会告知先等待，直到池中有进程结束，才会创建新的进程来执行这些请求。
下面介绍一下multiprocessing模块下的Pool类下的几个方法：
1.apply()
函数原型：apply(func[, args=()[, kwds={}]])
该函数用于传递不定参数，同python中的apply函数一致，主进程会被阻塞直到函数执行结束（不建议使用，并且3.x以后不在出现）。
2.apply_async
函数原型：apply_async(func[, args=()[, kwds={}[, callback=None]]])与apply()用法一致，但它是非阻塞的且支持结果返回后进行回调。
3.map()
函数原型：map(func, iterable[, chunksize=None])
Pool类中的map方法，与内置的map函数用法行为基本一致，它会使进程阻塞直到结果返回。
注意：虽然第二个参数是一个迭代器，但在实际使用中，必须在整个队列都就绪后，程序才会运行子进程。
4.map_async()
函数原型：map_async(func, iterable[, chunksize[, callback]])
与map用法一致，但是它是非阻塞的。其有关事项见apply_async。
5.close()
关闭进程池pool，使其不在接受新的任务。
6.terminal()
结束工作进程，不在处理未处理的任务。
7.join()
主进程阻塞等待子进程的退出， join方法要在close或terminate之后使用。
下面我们看一个简单的multiprocessing.Pool类的例子：
from multiprocessing import Pool
from time import time, sleep
import random
def perform_task(id):
begin = time()
print("【{}号】子进程【开始】执行任务".format(id))
sleep(random.random() * 5)
print("【{}号】子进程【结束】执行任务".format(id))
end = time()
cost = end - begin
print("【{}号】子进程执行任务耗时：{}".format(id, cost))
if __name__ == "__main__":
pool_count = 5
pool = Pool(pool_count)
print("进程池准备就绪")
print("多进程开始执行任务，等待结束...")
for i in range(pool_count):
pool.apply_async(perform_task, args=(i,))
pool.close()
pool.join()
print("所有进程池中的任务完成")
运行结果：
进程池准备就绪
多进程开始执行任务，等待结束...
【0号】子进程【开始】执行任务
【1号】子进程【开始】执行任务
【2号】子进程【开始】执行任务
【3号】子进程【开始】执行任务
【4号】子进程【开始】执行任务
【0号】子进程【结束】执行任务
【0号】子进程执行任务耗时：0.3483567237854004
【3号】子进程【结束】执行任务
【3号】子进程执行任务耗时：1.0760245323181152
【4号】子进程【结束】执行任务
【4号】子进程执行任务耗时：1.3666496276855469
【1号】子进程【结束】执行任务
【1号】子进程执行任务耗时：3.833340644836426
【2号】子进程【结束】执行任务
【2号】子进程执行任务耗时：4.259188890457153
所有进程池中的任务完成
对Pool对象调用join()方法会等待所有子进程执行完毕，调用join()之前必须先调用close()，调用close()之后就不能继续添加新的Process了。
进程间通信
Process之间肯定是需要通信的，操作系统提供了很多机制来实现进程间的通信。Python的multiprocessing模块包装了底层的机制，提供了Queue、Pipes等多种方式来交换数据。
我们以Queue为例，在父进程中创建两个子进程，一个往Queue里写数据，一个从Queue里读数据：
from multiprocessing import Process, Queue
import os, time, random
# 写数据进程执行的代码:
def write(q):
print('Process to write: %s' % os.getpid())
for value in ['A', 'B', 'C']:
print('Put %s to queue...' % value)
q.put(value)
time.sleep(random.random())
# 读数据进程执行的代码:
def read(q):
print('Process to read: %s' % os.getpid())
while True:
value = q.get(True)
print('Get %s from queue.' % value)
if __name__=='__main__':
# 父进程创建Queue，并传给各个子进程：
q = Queue()
pw = Process(target=write, args=(q,))
pr = Process(target=read, args=(q,))
# 启动子进程pw，写入:
pw.start()
# 启动子进程pr，读取:
pr.start()
# 等待pw结束:
pw.join()
# pr进程里是死循环，无法等待其结束，只能强行终止:
pr.terminate()
运行结果：
Process to write: 23192
Put A to queue...
Process to read: 2824
Get A from queue.
Put B to queue...
Get B from queue.
Put C to queue...
Get C from queue.
小结
本文主要学习了python中的多进程Process及实现跨平台的多进程multiprocessing，进程池Pool、通过Queue进程间通信。
更多了解，可关注公众号：人人懂编程
微信公众号：人人懂编程
网友评论