进程

Python 中的多线程其实并不是真正的多线程，如果想要充分地使用多核 CPU 的资源，在 Python 中大部分情况需要使用多进程。Python 提供了非常好用的多进程包 multiprocessing，只需要定义一个函数，Python 会完成其他所有事情。借助这个包，可以轻松完成从单进程到并发执行的转换。multiprocessing 支持子进程、通信和共享数据、执行不同形式的同步，提供了 Process、Queue、Pipe、Lock 等组件。

1、类 Process

创建进程的类：Process([group [, target [, name [, args [, kwargs]]]]])

  • target 表示调用对象
  • args 表示调用对象的位置参数元组
  • kwargs表示调用对象的字典
  • name为别名
  • group实质上不使用

下面看一个创建函数并将其作为多个进程的例子：

  1. #!/usr/bin/env python3
  2. # -*- coding: UTF-8 -*-
  4. import multiprocessing
  5. import time
  8. def worker(interval, name):
  9.     print(name + '【start】')
  10.     time.sleep(interval)
  11.     print(name + '【end】')
  14. if __name__ == "__main__":
  15.     p1 = multiprocessing.Process(target=worker, args=(2, '两点水1'))
  16.     p2 = multiprocessing.Process(target=worker, args=(3, '两点水2'))
  17.     p3 = multiprocessing.Process(target=worker, args=(4, '两点水3'))
  19.     p1.start()
  20.     p2.start()
  21.     p3.start()
  23.     print("The number of CPU is:" + str(multiprocessing.cpu_count()))
  24.     for p in multiprocessing.active_children():
  25.         print("child   p.name:" + p.name + "\tp.id" + str(p.pid))
  26.     print("END!!!!!!!!!!!!!!!!!")

输出的结果：

多进程输出结果

2、把进程创建成类

当然我们也可以把进程创建成一个类，如下面的例子，当进程 p 调用 start() 时，自动调用 run() 方法。

  1. # -*- coding: UTF-8 -*-
  3. import multiprocessing
  4. import time
  7. class ClockProcess(multiprocessing.Process):
  8.     def __init__(self, interval):
  9.         multiprocessing.Process.__init__(self)
  10.         self.interval = interval
  12.     def run(self):
  13.         n = 5
  14.         while n > 0:
  15.             print("当前时间: {0}".format(time.ctime()))
  16.             time.sleep(self.interval)
  17.             n -= 1
  20. if __name__ == '__main__':
  21.     p = ClockProcess(3)
  22.     p.start()

输出结果如下：

创建进程类

3、daemon 属性

想知道 daemon 属性有什么用，看下下面两个例子吧，一个加了 daemon 属性，一个没有加，对比输出的结果：

没有加 deamon 属性的例子：

  1. # -*- coding: UTF-8 -*-
  2. import multiprocessing
  3. import time
  6. def worker(interval):
  7.     print('工作开始时间：{0}'.format(time.ctime()))
  8.     time.sleep(interval)
  9.     print('工作结果时间：{0}'.format(time.ctime()))
  12. if __name__ == '__main__':
  13.     p = multiprocessing.Process(target=worker, args=(3,))
  14.     p.start()
  15.     print('【EMD】')

输出结果：

  1. 【EMD】
  2. 工作开始时间：Mon Oct  9 17:47:06 2017
  3. 工作结果时间：Mon Oct  9 17:47:09 2017

在上面示例中，进程 p 添加 daemon 属性：

  1. # -*- coding: UTF-8 -*-
  3. import multiprocessing
  4. import time
  7. def worker(interval):
  8.     print('工作开始时间：{0}'.format(time.ctime()))
  9.     time.sleep(interval)
  10.     print('工作结果时间：{0}'.format(time.ctime()))
  13. if __name__ == '__main__':
  14.     p = multiprocessing.Process(target=worker, args=(3,))
  15.     p.daemon = True
  16.     p.start()
  17.     print('【EMD】')

输出结果：

  1. 【EMD】

根据输出结果可见，如果在子进程中添加了 daemon 属性，那么当主进程结束的时候，子进程也会跟着结束。所以没有打印子进程的信息。

4、join 方法

结合上面的例子继续，如果我们想要让子线程执行完该怎么做呢？

那么我们可以用到 join 方法，join 方法的主要作用是：阻塞当前进程，直到调用 join 方法的那个进程执行完，再继续执行当前进程。

因此看下加了 join 方法的例子：

  1. import multiprocessing
  2. import time
  5. def worker(interval):
  6.     print('工作开始时间：{0}'.format(time.ctime()))
  7.     time.sleep(interval)
  8.     print('工作结果时间：{0}'.format(time.ctime()))
  11. if __name__ == '__main__':
  12.     p = multiprocessing.Process(target=worker, args=(3,))
  13.     p.daemon = True
  14.     p.start()
  15.     p.join()
  16.     print('【EMD】')

输出的结果：

  1. 工作开始时间：Tue Oct 10 11:30:08 2017
  2. 工作结果时间：Tue Oct 10 11:30:11 2017
  3. 【EMD】

5、Pool

如果需要很多的子进程，难道我们需要一个一个的去创建吗？

当然不用，我们可以使用进程池的方法批量创建子进程。

例子如下：

  1. # -*- coding: UTF-8 -*-
  3. from multiprocessing import Pool
  4. import os, time, random
  7. def long_time_task(name):
  8.     print('进程的名称：{0} ；进程的PID: {1} '.format(name, os.getpid()))
  9.     start = time.time()
  10.     time.sleep(random.random() * 3)
  11.     end = time.time()
  12.     print('进程 {0} 运行了 {1} 秒'.format(name, (end - start)))
  15. if __name__ == '__main__':
  16.     print('主进程的 PID：{0}'.format(os.getpid()))
  17.     p = Pool(4)
  18.     for i in range(6):
  19.         p.apply_async(long_time_task, args=(i,))
  20.     p.close()
  21.     # 等待所有子进程结束后在关闭主进程
  22.     p.join()
  23.     print('【End】')

输出的结果如下：

  1. 主进程的 PID：7256
  2. 进程的名称：0 ；进程的PID: 1492
  3. 进程的名称：1 ；进程的PID: 12232
  4. 进程的名称：2 ；进程的PID: 4332
  5. 进程的名称：3 ；进程的PID: 11604
  6. 进程 2 运行了 0.6500370502471924 秒
  7. 进程的名称：4 ；进程的PID: 4332
  8. 进程 1 运行了 1.0830621719360352 秒
  9. 进程的名称：5 ；进程的PID: 12232
  10. 进程 5 运行了 0.029001712799072266 秒
  11. 进程 4 运行了 0.9720554351806641 秒
  12. 进程 0 运行了 2.3181326389312744 秒
  13. 进程 3 运行了 2.5331451892852783 秒
  14. 【End】

这里有一点需要注意： Pool 对象调用 join() 方法会等待所有子进程执行完毕，调用 join() 之前必须先调用 close() ，调用close() 之后就不能继续添加新的 Process 了。

请注意输出的结果，子进程 0，1，2，3是立刻执行的，而子进程 4 要等待前面某个子进程完成后才执行，这是因为 Pool 的默认大小在我的电脑上是 4，因此，最多同时执行 4 个进程。这是 Pool 有意设计的限制，并不是操作系统的限制。如果改成：

  1. p = Pool(5)

就可以同时跑 5 个进程。

6、进程间通信

Process 之间肯定是需要通信的，操作系统提供了很多机制来实现进程间的通信。Python 的 multiprocessing 模块包装了底层的机制，提供了Queue、Pipes 等多种方式来交换数据。

以 Queue 为例，在父进程中创建两个子进程，一个往 Queue 里写数据，一个从 Queue 里读数据：

  1. #!/usr/bin/env python3
  2. # -*- coding: UTF-8 -*-
  4. from multiprocessing import Process, Queue
  5. import os, time, random
  8. def write(q):
  9.     # 写数据进程
  10.     print('写进程的PID:{0}'.format(os.getpid()))
  11.     for value in ['两点水', '三点水', '四点水']:
  12.         print('写进 Queue 的值为：{0}'.format(value))
  13.         q.put(value)
  14.         time.sleep(random.random())
  17. def read(q):
  18.     # 读取数据进程
  19.     print('读进程的PID:{0}'.format(os.getpid()))
  20.     while True:
  21.         value = q.get(True)
  22.         print('从 Queue 读取的值为：{0}'.format(value))
  25. if __name__ == '__main__':
  26.     # 父进程创建 Queue，并传给各个子进程
  27.     q = Queue()
  28.     pw = Process(target=write, args=(q,))
  29.     pr = Process(target=read, args=(q,))
  30.     # 启动子进程 pw
  31.     pw.start()
  32.     # 启动子进程pr
  33.     pr.start()
  34.     # 等待pw结束:
  35.     pw.join()
  36.     # pr 进程里是死循环，无法等待其结束，只能强行终止
  37.     pr.terminate()

输出的结果为：

  1. 读进程的PID:13208
  2. 写进程的PID:10864
  3. 写进 Queue 的值为：两点水
  4. 从 Queue 读取的值为：两点水
  5. 写进 Queue 的值为：三点水
  6. 从 Queue 读取的值为：三点水
  7. 写进 Queue 的值为：四点水
  8. 从 Queue 读取的值为：四点水