multiprocessing 多进程之间相互协调的方式有如下几种:

Lock:锁,
Semaphore:信号量
Queue:队列,
Event:事件
Pipe:管道

Queue队列通信的补充

补充说明关于：

使用Queue进行进程间通讯，通过multiprocessing.Queue(maxsize)对象进行通信。

Queue遵循的是：FIFO即First in First Out,先进先出的原则。
Queue提供了一个基本的FIFO容器，使用方法很简单,maxsize是个整数，指明了队列中能存放的数据个数的上限。

PS：如果队列数已经达到上限，则新数据的插入会导致阻塞，直到队列中的其他数据被消费后才会进行插入。如果maxsize小于或者等于0，队列大小没有限制。

进程池

内容

进程池的同步和异步
进程池结果关注
进程池的结果回调
进程池的map函数

1.什么是进程池?

进程序列，是一组资源进程，对这一组进程统一管理。

2：进程池的特点？

流程：

1：定义一个序列池，创建出来固定数量的进程对象，有需求来了，就拿一个池中的进程来处理任务
2:当使用时一个进程对象的时候，则去进程池中获取一个进程对象出来进行任务处理
3:任务处理完毕，进程不退出（不进行销毁），是将进程再放回进程池中继续等待新的任务委派
4:如果进程池序列中没有可供使用的进进程，那么程序就会等待，直到进程池中有可用进程为止。

所以进程池的特点是：

【同一时间最多有多少个进程】能够【同时执行任务】和【进程池中】进程的【个数】有关系

如果进程池设置了个数，那么程序只会开启四个进程运行

2:为什么进程无限多开无意义且还会性能更慢？

1：创建进程和销毁进程都需要消耗时间
2：即便开启很多的进程，进程能分配到CPU上执行的也就只能是几个核的数量，操作系统并也不能这些进程同时执行，过多的进程反而会影响程序的效率，进行的上下文的切换也是需要耗时的。

3.进程池好处?

1）并发高效的执行任务
2）减少操作系统的调度难度，节省开关进程的时间，一定程度上能够实现并发效果。

4：多进程不适宜的场景：

1：非密集型CPU运算任务。如果是密集型IO不太合适（当然web服务类的可以开多进程，进行有效的多核的分配利用）

2:通常我们任务数和开启的进程数是成比例的，多开进程其实是会引起更多性能消耗，实际上并不划算。

3：通常超过了 cpu 个数的任务数，可以考虑使用进程池来解决问题,而不是无限多开启子进程方式进行解决。

进程池的同步和异步模式

1 进程池的常用的几个方法：

apply：同步执行（串行），需当前子进程执行完毕后，才执行下一个进程，一般不使用
apply_async：异步执行（并行），不需要等待当前子进程执行完毕，随时可以根据系统调度来进行任务执行。
terminate() ：立刻关闭进程池
join() ：主进程等待所有子进程执行完毕。必须在close或terminate()之后。
close() ：关闭进程池。需等待所有进程结束后，才可以关闭。

PS:如果Pool对象调用join()方法后会等待所有子进程执行完毕，调用join()之前必须先调用close()方法，调用close()之后则不能继续添加新的Process。

2 进程池同步模式（阻塞）：

PS：示例运行如果再pycharm无法运行，需要丢到dos下执行。

示例：


import time
from multiprocessing import Pool,current_process
def fun1(msg):
    print('子进程接受消息：%s'%(msg),"子进程的ID%s"%( current_process().pid))

    time.sleep(3)
    print('子进程执行完成')

if __name__ == '__main__':
    p = Pool(3)
    start = time.time()
    for i in range(10):
         p.apply(func=fun1,args=('我是编号：%s'%(i),))
    print("主进程执行完毕，总计耗时：%s"%(time.time()-start))

输出结果：

D:\ceshi>python jincheng11.py
子进程接受消息：我是编号：0 子进程的ID14476
子进程执行完成
子进程接受消息：我是编号：1 子进程的ID32092
子进程执行完成
子进程接受消息：我是编号：2 子进程的ID22924
子进程执行完成
子进程接受消息：我是编号：3 子进程的ID14476
子进程执行完成
子进程接受消息：我是编号：4 子进程的ID32092
子进程执行完成
子进程接受消息：我是编号：5 子进程的ID22924
子进程执行完成
子进程接受消息：我是编号：6 子进程的ID14476
子进程执行完成
子进程接受消息：我是编号：7 子进程的ID32092
子进程执行完成
子进程接受消息：我是编号：8 子进程的ID22924
子进程执行完成
子进程接受消息：我是编号：9 子进程的ID14476
子进程执行完成
主进程执行完毕，总计耗时：30.252848148345947

分析：

上面的同步模式中：

1：进程池的数量是3：但是我们的10个任务：所以任务领取到的进程ID是池中的同一个进程的ID。
2：且是同步机制，需要等待没一个子进程执行完成才可以下一步！
3：总耗时和3个池*10任务基本吻合。

3 进程池异步模式（非阻塞）：

PS：示例运行如果再pycharm无法运行，需要丢到dos下执行。

示例：

#!/usr/bin/evn python
# -*- coding: utf-8 -*-
"""
-------------------------------------------------
   文件名称 :     jincheng10
   文件功能描述 :   功能描述
   创建人 :       小钟同学
   创建时间 :          2021/5/7
-------------------------------------------------
   修改描述-2021/5/7:         
-------------------------------------------------
"""
import time
from multiprocessing import Pool,current_process
import os

def fun1(msg):
    print('子进程接受消息：%s'%(msg),"子进程的ID%s"%( current_process().pid))

    time.sleep(3)
    print('子进程执行完成',"子进程的ID%s"%( current_process().pid))

if __name__ == '__main__':
    print("父进程 (%s) 启动 (%s) 个cpu核" % (os.getpid(), os.cpu_count()))
    p = Pool(3)
    start = time.time()
    for i in range(10):
         p.apply_async(func=fun1,args=('我是编号：%s'%(i),))
         # p.join()
    p.close()  # 关闭进程池，关闭后则进程池对象不再接受新的请求
    p.join()  # 等待进程池p中的所有子进程执行完成后主进程才退出，必须放在close语句之后,
    print("主进程执行完毕，总计耗时：%s"%(time.time()-start))

输出结果：

D:\ceshi>python jincheng11.py
父进程 (30364) 启动 (6) 个cpu核
子进程接受消息：我是编号：0 子进程的ID41792
子进程接受消息：我是编号：1 子进程的ID38072
子进程接受消息：我是编号：2 子进程的ID43004
子进程执行完成 子进程的ID41792
子进程接受消息：我是编号：3 子进程的ID41792
子进程执行完成 子进程的ID38072
子进程接受消息：我是编号：4 子进程的ID38072
子进程执行完成 子进程的ID43004
子进程接受消息：我是编号：5 子进程的ID43004
子进程执行完成 子进程的ID41792
子进程接受消息：我是编号：6 子进程的ID41792
子进程执行完成 子进程的ID38072
子进程接受消息：我是编号：7 子进程的ID38072
子进程执行完成 子进程的ID43004
子进程接受消息：我是编号：8 子进程的ID43004
子进程执行完成 子进程的ID41792
子进程接受消息：我是编号：9 子进程的ID41792
子进程执行完成 子进程的ID38072
子进程执行完成 子进程的ID43004
子进程执行完成 子进程的ID41792
主进程执行完毕，总计耗时：12.333166122436523

D:\ceshi>

3 进程池异步模式（非阻塞）+ 关注我们的异步执行结果：

PS：示例运行如果再pycharm无法运行，需要丢到dos下执行。

进程池里面可以使用 get() 获取func() 的返回值

示例：

import time
from multiprocessing import Pool,current_process
import os

def fun1(msg):
    print('子进程接受消息：%s'%(msg),"子进程的ID%s"%( current_process().pid))

    time.sleep(3)
    print('子进程执行完成')
    return "子进程执行完成了：%s"%(current_process().pid)

if __name__ == '__main__':
    print("父进程 (%s) 启动 (%s) 个cpu核" % (os.getpid(), os.cpu_count()))
    p = Pool(3)
    result_ps= []
    start = time.time()
    for i in range(10):
        result_ps.append(p.apply_async(func=fun1,args=('我是编号：%s'%(i),)))
         # p.join()
    p.close()  # 关闭进程池，关闭后则进程池对象不再接受新的请求
    p.join()  # 等待进程池p中的所有子进程执行完成后主进程才退出，必须放在close语句之后,

    #查询我们的执行结果：
    for item in result_ps:
        #获取每个进程执行的结果
        print(item.get())


    print("主进程执行完毕，总计耗时：%s"%(time.time()-start))

输出结果：

D:\ceshi>python jincheng11.py
父进程 (6632) 启动 (6) 个cpu核
子进程接受消息：我是编号：0 子进程的ID22260
子进程接受消息：我是编号：1 子进程的ID34036
子进程接受消息：我是编号：2 子进程的ID37216
子进程执行完成
子进程接受消息：我是编号：3 子进程的ID22260
子进程执行完成
子进程接受消息：我是编号：4 子进程的ID34036
子进程执行完成
子进程接受消息：我是编号：5 子进程的ID37216
子进程执行完成
子进程接受消息：我是编号：6 子进程的ID22260
子进程执行完成
子进程接受消息：我是编号：7 子进程的ID34036
子进程执行完成
子进程接受消息：我是编号：8 子进程的ID37216
子进程执行完成
子进程接受消息：我是编号：9 子进程的ID22260
子进程执行完成
子进程执行完成
子进程执行完成
子进程执行完成了：22260
子进程执行完成了：34036
子进程执行完成了：37216
子进程执行完成了：22260
子进程执行完成了：34036
子进程执行完成了：37216
子进程执行完成了：22260
子进程执行完成了：34036
子进程执行完成了：37216
子进程执行完成了：22260
主进程执行完毕，总计耗时：12.287627458572388

4 进程池异步模式（非阻塞）+ 执行回调：

PS：示例运行如果再pycharm无法运行，需要丢到dos下执行。

回调函数是执行在主进程的下:

示例：

import time
from multiprocessing import Pool,current_process
import os

def fun1(msg):
    print('子进程接受消息：%s'%(msg),"子进程的ID%s"%( current_process().pid))

    time.sleep(3)
    print('子进程执行完成')
    return "子进程执行完成了：%s"%(current_process().pid)


def funl_callback(msg):
    pass
    # 回调函数用来保证一个任务在执行完毕之后，立刻发出 callback 回调函数中的内容 并且子进程的函数返回值作为 callback 函数的参数
    print("子进程%s完成了函数回调！回调结果是：%s：" % (current_process().pid,msg))
    print("==="*30)



if __name__ == '__main__':
    print("父进程 (%s) 启动 (%s) 个cpu核" % (os.getpid(), os.cpu_count()))
    p = Pool(3)

    start = time.time()
    for i in range(10):
        p.apply_async(func=fun1, args=('我是编号：%s' % (i),), callback=funl_callback)
         # p.join()
    p.close()  # 关闭进程池，关闭后则进程池对象不再接受新的请求
    p.join()  # 等待进程池p中的所有子进程执行完成后主进程才退出，必须放在close语句之后,



    print("主进程执行完毕，总计耗时：%s"%(time.time()-start))

输出结果为：

mayn@DESKTOP-16CKEN1 MINGW64 /d/ceshi
$ python jincheng11.py
父进程 (38156) 启动 (6) 个cpu核
子进程接受消息：我是编号：0 子进程的ID39812
子进程接受消息：我是编号：1 子进程的ID34344
子进程接受消息：我是编号：2 子进程的ID41352
子进程执行完成
子进程接受消息：我是编号：3 子进程的ID39812
子进程38156完成了函数回调！回调结果是：子进程执行完成了：39812：
==========================================================================================
子进程执行完成
子进程38156完成了函数回调！回调结果是：子进程执行完成了：34344：
子进程接受消息：我是编号：4 子进程的ID34344
==========================================================================================
子进程执行完成
子进程接受消息：我是编号：5 子进程的ID41352
子进程38156完成了函数回调！回调结果是：子进程执行完成了：41352：
==========================================================================================
子进程执行完成
子进程38156完成了函数回调！回调结果是：子进程执行完成了：39812：
子进程接受消息：我是编号：6 子进程的ID39812
==========================================================================================
子进程执行完成
子进程接受消息：我是编号：7 子进程的ID34344
子进程38156完成了函数回调！回调结果是：子进程执行完成了：34344：
==========================================================================================
子进程执行完成
子进程38156完成了函数回调！回调结果是：子进程执行完成了：41352：
子进程接受消息：我是编号：8 子进程的ID41352
==========================================================================================
子进程执行完成
子进程接受消息：我是编号：9 子进程的ID39812
子进程38156完成了函数回调！回调结果是：子进程执行完成了：39812：
==========================================================================================
子进程执行完成
子进程38156完成了函数回调！回调结果是：子进程执行完成了：34344：
==========================================================================================
子进程执行完成
子进程38156完成了函数回调！回调结果是：子进程执行完成了：41352：
==========================================================================================
子进程执行完成
子进程38156完成了函数回调！回调结果是：子进程执行完成了：39812：
==========================================================================================
主进程执行完毕，总计耗时：12.317128896713257

mayn@DESKTOP-16CKEN1 MINGW64 /d/ceshi

5 进程池中的map方法：

PS：示例运行如果再pycharm无法运行，需要丢到dos下执行。

map()函数：和我们的常规高阶函数map用法一样。

第一个参数就是需要引用的函数
第二个参数是一个可迭代对象，它会把需要迭代的元素一个个的传入第一个参数我们的函数。

PS：但是里面使用的还是一个进程来执行运行

示例

import time
from multiprocessing.pool import Pool
from multiprocessing import current_process
def numsCheng(i):
    print('子进程接受消息：%s' % (i), "子进程的ID%s" % (current_process().pid))
    return i * 2

if __name__ == '__main__':
    time1 = time.time()
    nums_list = [1, 2, 3, 4, 5, 6, 7, 8, 9]
    pool = Pool(processes=5)
    result = pool.map(numsCheng, nums_list)
    pool.close()        # 关闭进程池，不再接受新的进程
    pool.join()         # 主进程阻塞等待子进程的退出

    print(result)
    time2 = time.time()
    print("计算用时：", time2-time1)

输出结果为：

D:\ceshi>python jincheng12.py
子进程接受消息：1 子进程的ID36944
子进程接受消息：2 子进程的ID36944
子进程接受消息：3 子进程的ID36944
子进程接受消息：4 子进程的ID36944
子进程接受消息：5 子进程的ID36944
子进程接受消息：6 子进程的ID36944
子进程接受消息：7 子进程的ID36944
子进程接受消息：8 子进程的ID36944
子进程接受消息：9 子进程的ID36944
[2, 4, 6, 8, 10, 12, 14, 16, 18]
计算用时： 0.4478025436401367

个人其他博客地址

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

小钟同学 | 文【原创】| QQ：308711822

1：本文相关描述主要是个人的认知和见解，如有不当之处，还望各位大佬指正。

2：关于文章内容，有部分内容参考自互联网整理，如有链接会声明标注；如没有及时标注备注的链接的，如有侵权请联系，我会立即删除处理哟。

ITPub博客

python multiprocessing 多进程之进程池

Queue队列通信的补充

进程池

1.什么是进程池?

2：进程池的特点？

2:为什么进程无限多开无意义且还会性能更慢？

3.进程池好处?

4：多进程不适宜的场景：

进程池的同步和异步模式

1 进程池的常用的几个方法：

2 进程池同步模式（阻塞）：

3 进程池异步模式（非阻塞）：

3 进程池异步模式（非阻塞）+ 关注我们的异步执行结果：

4 进程池异步模式（非阻塞）+ 执行回调：

5 进程池中的map方法：

个人其他博客地址

最新文章