# 进程 # Python 中的多线程其实并不是真正的多线程，如果想要充分地使用多核 CPU 的资源，在 Python 中大部分情况需要使用多进程。 Python 提供了非常好用的多进程包 multiprocessing，只需要定义一个函数，Python 会完成其他所有事情。 借助这个包，可以轻松完成从单进程到并发执行的转换。multiprocessing 支持子进程、通信和共享数据、执行不同形式的同步，提供了 Process、Queue、Pipe、Lock 等组件。 ## 1、类 Process ## 创建进程的类：`Process([group [, target [, name [, args [, kwargs]]]]])` * target 表示调用对象 * args 表示调用对象的位置参数元组 * kwargs表示调用对象的字典 * name为别名 * group实质上不使用 下面看一个创建函数并将其作为多个进程的例子： ```python #!/usr/bin/env python3 # -*- coding: UTF-8 -*- import multiprocessing import time def worker(interval, name): print(name + '【start】') time.sleep(interval) print(name + '【end】') if __name__ == "__main__": p1 = multiprocessing.Process(target=worker, args=(2, '两点水1')) p2 = multiprocessing.Process(target=worker, args=(3, '两点水2')) p3 = multiprocessing.Process(target=worker, args=(4, '两点水3')) p1.start() p2.start() p3.start() print("The number of CPU is:" + str(multiprocessing.cpu_count())) for p in multiprocessing.active_children(): print("child p.name:" + p.name + "\tp.id" + str(p.pid)) print("END!!!!!!!!!!!!!!!!!") ``` 输出的结果： ``` The number of CPU is:4 child p.name:Process-3 p.id7600 child p.name:Process-2 p.id5560 child p.name:Process-1 p.id7716 END!!!!!!!!!!!!!!!!! 两点水 1【start】 两点水 2【start】 两点水 3【start】 两点水 1【end】 两点水 2【end】 两点水 3【end】 Process finished with exit code 0 ``` ## 2、把进程创建成类 ## 当然我们也可以把进程创建成一个类，如下面的例子，当进程 p 调用 start() 时，自动调用 run() 方法。 ```python # -*- coding: UTF-8 -*- import multiprocessing import time class ClockProcess(multiprocessing.Process): def __init__(self, interval): multiprocessing.Process.__init__(self) self.interval = interval def run(self): n = 5 while n > 0: print(f"当前时间: {time.ctime()}") time.sleep(self.interval) n -= 1 if __name__ == '__main__': p = ClockProcess(3) p.start() ``` 输出结果如下： ``` 当前时间: Mon Oct 9 17:26:16 2017 当前时间: Mon Oct 9 17:26:19 2017 当前时间: Mon Oct 9 17:26:22 2017 当前时间: Mon Oct 9 17:26:25 2017 当前时间: Mon Oct 9 17:26:28 2017 Process finished with exit code 0 ``` ## 3、daemon 属性 ## 想知道 daemon 属性有什么用，看下下面两个例子吧，一个加了 daemon 属性，一个没有加，对比输出的结果： 没有加 deamon 属性的例子： ```python # -*- coding: UTF-8 -*- import multiprocessing import time def worker(interval): print(f'工作开始时间：{time.ctime()}') time.sleep(interval) print(f'工作结果时间：{time.ctime()}') if __name__ == '__main__': p = multiprocessing.Process(target=worker, args=(3,)) p.start() print('【EMD】') ``` 输出结果： ```txt 【EMD】 工作开始时间：Mon Oct 9 17:47:06 2017 工作结果时间：Mon Oct 9 17:47:09 2017 ``` 在上面示例中，进程 p 添加 daemon 属性： ```python # -*- coding: UTF-8 -*- import multiprocessing import time def worker(interval): print(f'工作开始时间：{time.ctime()}') time.sleep(interval) print(f'工作结果时间：{time.ctime()}') if __name__ == '__main__': p = multiprocessing.Process(target=worker, args=(3,)) p.daemon = True p.start() print('【EMD】') ``` 输出结果： ```txt 【EMD】 ``` 根据输出结果可见，如果在子进程中添加了 daemon 属性，那么当主进程结束的时候，子进程也会跟着结束。所以没有打印子进程的信息。 ## 4、join 方法 ## 结合上面的例子继续，如果我们想要让子线程执行完该怎么做呢？ 那么我们可以用到 join 方法，join 方法的主要作用是：阻塞当前进程，直到调用 join 方法的那个进程执行完，再继续执行当前进程。 因此看下加了 join 方法的例子： ```python import multiprocessing import time def worker(interval): print(f'工作开始时间：{time.ctime()}') time.sleep(interval) print(f'工作结果时间：{time.ctime()}') if __name__ == '__main__': p = multiprocessing.Process(target=worker, args=(3,)) p.daemon = True p.start() p.join() print('【EMD】') ``` 输出的结果： ```txt 工作开始时间：Tue Oct 10 11:30:08 2017 工作结果时间：Tue Oct 10 11:30:11 2017 【EMD】 ``` ## 5、Pool ## 如果需要很多的子进程，难道我们需要一个一个的去创建吗？ 当然不用，我们可以使用进程池的方法批量创建子进程。 例子如下： ```python # -*- coding: UTF-8 -*- from multiprocessing import Pool import os, time, random def long_time_task(name): print(f'进程的名称：{name} ；进程的PID: {os.getpid()} ') start = time.time() time.sleep(random.random() * 3) end = time.time() print(f'进程 {name} 运行了 {end - start} 秒') if __name__ == '__main__': print(f'主进程的 PID：{os.getpid()}') p = Pool(4) for i in range(6): p.apply_async(long_time_task, args=(i,)) p.close() # 等待所有子进程结束后在关闭主进程 p.join() print('【End】') ``` 输出的结果如下： ```txt 主进程的 PID：7256 进程的名称：0 ；进程的PID: 1492 进程的名称：1 ；进程的PID: 12232 进程的名称：2 ；进程的PID: 4332 进程的名称：3 ；进程的PID: 11604 进程 2 运行了 0.6500370502471924 秒 进程的名称：4 ；进程的PID: 4332 进程 1 运行了 1.0830621719360352 秒 进程的名称：5 ；进程的PID: 12232 进程 5 运行了 0.029001712799072266 秒 进程 4 运行了 0.9720554351806641 秒 进程 0 运行了 2.3181326389312744 秒 进程 3 运行了 2.5331451892852783 秒 【End】 ``` 这里有一点需要注意： `Pool` 对象调用 `join()` 方法会等待所有子进程执行完毕，调用 `join()` 之前必须先调用 `close()` ，调用`close()` 之后就不能继续添加新的 Process 了。 请注意输出的结果，子进程 0，1，2，3是立刻执行的，而子进程 4 要等待前面某个子进程完成后才执行，这是因为 Pool 的默认大小在我的电脑上是 4，因此，最多同时执行 4 个进程。这是 Pool 有意设计的限制，并不是操作系统的限制。如果改成： ```python p = Pool(5) ``` 就可以同时跑 5 个进程。 ## 6、进程间通信 ## Process 之间肯定是需要通信的，操作系统提供了很多机制来实现进程间的通信。Python 的 multiprocessing 模块包装了底层的机制，提供了Queue、Pipes 等多种方式来交换数据。 以 Queue 为例，在父进程中创建两个子进程，一个往 Queue 里写数据，一个从 Queue 里读数据： ```python #!/usr/bin/env python3 # -*- coding: UTF-8 -*- from multiprocessing import Process, Queue import os, time, random def write(q): # 写数据进程 print(f'写进程的PID:{os.getpid()}') for value in ['两点水', '三点水', '四点水']: print(f'写进 Queue 的值为：{value}') q.put(value) time.sleep(random.random()) def read(q): # 读取数据进程 print(f'读进程的PID:{os.getpid()}') while True: value = q.get(True) print(f'从 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() ``` 输出的结果为： ```txt 读进程的PID:13208 写进程的PID:10864 写进 Queue 的值为：两点水 从 Queue 读取的值为：两点水 写进 Queue 的值为：三点水 从 Queue 读取的值为：三点水 写进 Queue 的值为：四点水 从 Queue 读取的值为：四点水 ```