多进程和多线程的区别

  • Python多线程的操作，由于有GIL锁的存在，使得其运行效率并不会很高，无法充分利用 多核cpu 的优势，只有在I/O密集形的任务逻辑中才能实现并发。
  • 使用多进程来编写同样消耗cpu（一般是计算）的逻辑，对于 多核cpu 来说效率会好很多。
  • 操作系统对进程的调度代价要比线程调度要大的多。

多线程和多进程使用案例对比

1.用多进程和多线程两种方式来运算 斐波那契数列，这里都依赖 concurrent.futures 模块提供的线/进程池。

  1. import time
  2. from concurrent.futures import ThreadPoolExecutor
  3. from concurrent.futures import ProcessPoolExecutor
  4. from concurrent.futures import as_completed
  6. def fib(n):
  7.     return 1 if n <= 2 else fib(n-1) + fib(n-2)
  9. if __name__ == '__main__':
  10.     # with ProcessPoolExecutor(3) as executor:
  11.     with ThreadPoolExecutor(3) as executor:
  12.         all_task = [executor.submit(fib, n) for n in range(25, 35)]
  13.         start_time = time.time()
  14.         for future in as_completed(all_task):
  15.             data = future.result()
  16.             # todo
  18.         end_time = time.time()
  21.     print("time consuming by threads: {0}s".format(end_time-start_time))
  22.     # print("time consuming by processes: {0}s".format(end_time-start_time))

两种方式的运行结果对比：

  1. # result:
  2. # time consuming by threads: 4.823292016983032s
  3. # time consuming by processes: 3.3890748023986816s

可以看到，对于高计算量的任务，多进程要比多线程更加高效。同时，从这个例子中还能看出，通过concurrent.futures模块使用线程池和进程池的方式的接口和使用逻辑是一样的，不过在使用多进程时，对于Windows的操作平台，相关逻辑一定要放在main中，Linux不受约束。

2.用多进程和多线程两种方式来模拟 I/O密集操作，I/O操作 的特点就是 cpu 要耗费大量的时间进行等待数据，这里用sleep()进行模拟即可。

整体的操作方式不变，修改过的逻辑如下：

  1. def random_sleep(n):
  2.     time.sleep(n)
  3.     return n
  5. ...
  6. # 8 个线程，每个休眠两秒，模拟 I/O
  7. with ProcessPoolExecutor(8) as executor:
  8. # with ThreadPoolExecutor(8) as executor:
  9.     all_task = [executor.submit(random_sleep, 2) for i in range(30)]
  1. # result:
  2. # time consuming by threads: 8.002903699874878s
  3. # time consuming by processes: 8.34946894645691s

多进程编程

直接使用

  1. import time
  2. import multiprocessing
  5. def read(times):
  6.     time.sleep(times)
  7.     print("process reading...")
  8.     return "read for {0}s".format(times)
  10. def write(times):
  11.     time.sleep(times)
  12.     print("process writing...")
  13.     return "write for {0}s".format(times)
  15. if __name__ == '__main__':
  16.     read_process = multiprocessing.Process(target=read, args=(1,))
  17.     write_process = multiprocessing.Process(target=write, args=(2,))
  19.     read_process.start()
  20.     write_process.start()
  22.     print("read_process id {rid}".format(rid=read_process.pid))
  23.     print("write_process id {wid}".format(wid=write_process.pid))
  25.     read_process.join()
  26.     write_process.join()
  28.     print("done")
  30. # result:
  31. # read_process id 7064
  32. # write_process id 836
  33. # process reading...
  34. # process writing...
  35. # done

可以看出，关于多线程的逻辑和多线程的使用方式以类似的，要注意在Windows操作系统上，和进程有关的逻辑要写在if __name__ == '__main__'中。其他的一些方法请参阅 官方文档。

使用原生进程池

  1. import time
  2. import multiprocessing
  5. def read(times):
  6.     time.sleep(times)
  7.     print("process reading...")
  8.     return "read for {0}s".format(times)
  10. def write(times):
  11.     time.sleep(times)
  12.     print("process writing...")
  13.     return "write for {0}s".format(times)
  16. if __name__ == '__main__':
  17.     # multiprocessing.cpu_count() 获取cpu的核心数
  18.     pool = multiprocessing.Pool(multiprocessing.cpu_count())
  20.     read_result = pool.apply_async(read, args=(2,))
  21.     write_result = pool.apply_async(write, args=(3,))
  23.     # 关闭进程池，不再接受新的任务提交，否则 join() 出错
  24.     pool.close()
  25.     # 等待进程池中提交的所有任务完成
  26.     pool.join()
  29.     print(read_result.get())
  30.     print(write_result.get())
  33. # result:
  34. # process reading...
  35. # process writing...
  36. # read for 2s
  37. # write for 3s

使用imap()，所有任务顺序执行:

  1. pool = multiprocessing.Pool(multiprocessing.cpu_count())
  3. for result in pool.imap(read, [2, 1, 3]):
  4.     print(result)
  6. # result:
  7. # process reading...
  8. # process reading...
  9. # read for 2s
  10. # read for 1s
  11. # process reading...
  12. # read for 3s

使用imap_unordered()，哪个任务先完成就先返回结果：

  1. for result in pool.imap_unordered(read, [1, 5, 3]):
  2.     print(result)
  4. # process reading...
  5. # read for 1s
  6. # process reading...
  7. # read for 3s
  8. # process reading...
  9. # read for 5s

使用concurrent.futures中的ProcessPoolExecutor

这个在多线程和多进程对比的时提到过，因为和多线程的使用方式一样，这里就不多赘述，可以参阅 官方文档 给出的例子

进程间通信

进程通信和线程通信有些区别，在线程通信中各种提供的锁的机制和全局变量在这里不再适用，我们要选取新的工具来完成进程通信任务。

使用multiprocessing.Queue

使用逻辑是和多线程中的Queue是一样的，详细方法。这种通信方式不能用在通过Pool进程池创建的进程中

  1. import multiprocessing
  2. import time
  4. def plus(queue):
  5.     for i in range(6):
  6.         num = queue.get() + 1
  7.         queue.put(num)
  8.         print(num)
  9.         time.sleep(1)
  11. def subtract(queue):
  12.     for i in range(6):
  13.         num = queue.get() - 1
  14.         queue.put(num)
  15.         print(num)
  16.         time.sleep(2)
  18. if __name__ == '__main__':
  19.     queue = multiprocessing.Queue(1)
  20.     queue.put(0)
  22.     plus_process = multiprocessing.Process(target=plus, args=(queue,))
  23.     subtract_process = multiprocessing.Process(target=subtract, args=(queue,))
  25.     plus_process.start()
  26.     subtract_process.start()
  28. # result:
  29. # 1
  30. # 1
  31. # 2
  32. # 2
  33. # 3
  34. # 3
  35. # 0
  36. # 1
  37. # 2
  38. # 2
  39. # 1
  40. # 0

使用Manager()中的Queue

Manager()会返回一个在进程间进行同步管理的一个对象，它提供了多种在进程间共享数据的形式。

  1. import multiprocessing
  2. import time
  4. def plus(queue):
  5.     for i in range(6):
  6.         num = queue.get() + 1
  7.         queue.put(num)
  8.         print(num)
  9.         time.sleep(1)
  11. def subtract(queue):
  12.     for i in range(6):
  13.         num = queue.get() - 1
  14.         queue.put(num)
  15.         print(num)
  16.         time.sleep(2)
  18. if __name__ == '__main__':
  19.     queue = multiprocessing.Manager().Queue(1)      # 创建方式有些奇特
  20.     # queue = multiprocessing.Queue()     # 这时用这个就行不通了
  21.     pool = multiprocessing.Pool(2)
  23.     queue.put(0)
  25.     pool.apply_async(plus, args=(queue,))
  26.     pool.apply_async(subtract, args=(queue,))
  28.     pool.close()
  29.     pool.join()
  31. # result:
  32. # 0
  33. # 1
  34. # 1
  35. # 2
  36. # 2
  37. # 3
  38. # -1
  39. # 0
  40. # 1
  41. # 2
  42. # 1
  43. # 0

使用Manager()中的list()

多个进程可以共享全局的list，因为是进程间共享，所以用锁的机制保证它的安全性。这里的Manager().Lock不是前面线程级别的Lock，它可以保证进程间的同步。

  1. import multiprocessing as mp
  2. import time
  4. def add_person(waiting_list, name_list, lock):
  5.     lock.acquire()
  6.     for name in name_list:
  7.         waiting_list.append(name)
  8.         time.sleep(1)
  9.         print(waiting_list)
  10.     lock.release()
  12. def get_person(waiting_list, lock):
  13.     lock.acquire()
  14.     if waiting_list:
  15.         name = waiting_list.pop(0)
  16.         print("get {0}".format(name))
  17.     lock.release()
  20. if __name__ == '__main__':
  21.     waiting_list = mp.Manager().list()
  22.     lock = mp.Manager().Lock()      # 使用 lock 限制进程对全局量的访问
  23.     name_list = ["MetaTian", "Rity", "Anonymous"]
  25.     add_process = mp.Process(target=add_person, args=(waiting_list, name_list, lock))
  26.     get_process = mp.Process(target=get_person, args=(waiting_list, lock))
  28.     add_process.start()
  29.     get_process.start()
  31.     add_process.join()
  32.     get_process.join()
  34.     print(waiting_list)
  37. # result:
  38. # ['MetaTian']
  39. # ['MetaTian', 'Rity']
  40. # ['MetaTian', 'Rity', 'Anonymous']
  41. # get MetaTian
  42. # ['Rity', 'Anonymous']

Manager()中还有更多的进程间通信的工具，可以参阅官方文档。

使用Pipe

Pipe只能适用于两个进程间的通信，它的性能高于Queue，Pipe()会返回两个Connection对象，使用这个对象可以在进程间进行数据的发送和接收，非常像前面讲过的socket对象。关于Connection

  1. import multiprocessing
  3. def plus(conn):
  4.     default_num = 0
  5.     for i in range(3):
  6.         num = 0 if i == 0 else conn.recv()
  7.         conn.send(num + 1)
  8.         print("plus send: {0}".format(num+1))
  10. def subtract(conn):
  11.     for i in range(3):
  12.         num = conn.recv()
  13.         conn.send(num-1)
  14.         print("subtract send: {0}".format(num-1))
  16. if __name__ == '__main__':
  17.     conn_plus, conn_sbtract = multiprocessing.Pipe()
  19.     plus_process = multiprocessing.Process(target=plus, args=(conn_plus,))
  20.     subtract_process = multiprocessing.Process(target=subtract, args=(conn_sbtract,))
  22.     plus_process.start()
  23.     subtract_process.start()
  25. # result:
  26. # plus send: 1
  27. # subtract send: 0
  28. # plus send: 1
  29. # subtract send: 0
  30. # plus send: 1
  31. # subtract send: 0

send()可以连续发送数据，recv()将另一端发送的数据陆续取出，如果没有取到数据，则进入等待状态。