进程池

Posted 2021-01-07 qq752059037

进程池(Pool)

当需要创建的子进程数量不多时，可以直接利用multiprocessing中的Process动态成生多个进程，但如果是上百甚至上千个目标，手动的去创建进程的工作量巨大，此时就可以用到multiprocessing模块提供的Pool方法。

初始化Pool时，可以指定一个最大进程数，当有新的请求提交到Pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到指定的最大值，那么该请求就会等待，直到池中有进程结束，才会用之前的进程来执行新的任务

一般不会创建太多进程,因为CPU工作量太大,我们一般根据CPU个数创建,CPU个数可以os.cpu_count()获得

import time
from multiprocessing import Pool,Process
def func(num):
    time.sleep(0.2)
    print(‘%s号进程完毕‘%num)
#
if __name__ == ‘__main__‘:
    start = time.time()
    p = Pool(32) #这里指定最大进程数
    for i in range(100):
        p.apply_async(func,args=(i,))  # 异步提交func到一个子进程中执行
    p.close()    # 关闭进程池，用户不能再向这个池中提交任务了
    p.join()     # 阻塞，直到进程池中所有的任务都被执行完
    print(time.time() - start)
#这里我用1个进程:20.7秒 4个进程:6.3秒 8个进程:4.62秒 16个进程:5.6秒 32个进程:7.68秒

注意: 一定要关闭进程池然后join 因为主程序和子程序是完全异步的,不然子程序还没开启,主程序就结束了

---

 

multiprocessing.Pool常用函数解析：

• apply_async(func[, args[, kwds]]) ：使用非阻塞方式调用func（并行执行，堵塞方式必须等待上一个进程退出才能执行下一个进程），args为传递给func的参数列表，kwds为传递给func的关键字参数列表；
• close()：关闭Pool，使其不再接受新的任务；
• terminate()：不管任务是否完成，立即终止；
• join()：主进程阻塞，等待子进程的退出， 必须在close或terminate之后使用；

 

 

提交任务 (获取返回值)

提交任务
    同步提交 apply
        返回值 ： 子进程对应函数的返回值
        一个一个顺序执行的，并没有任何并发效果
    异步提交 apply_async
        没有返回值，要想所有任务能够顺利的执行完毕
            p.close()
            p.join() # 必须先close再join，阻塞直到进程池中的所有任务都执行完毕
        有返回值的情况下
            res.get() # get不能再提交任务之后立刻执行，应该是先提交所有的任务再通过get获取结果
        map()方法
            异步提交的简化版本
            自带close和join方法

 

使用apply同步提交

def task(num):
    time.sleep(1)
    print(‘%s : %s‘%(num,os.getpid()))
    return num**2
 
if __name__ == ‘__main__‘:
    p = Pool()
    for i in range(20):
        res = p.apply(task,args=(i,))   # 同步提交,必须等子程序执行完然后打印 才会进行下一个进程
        print(‘-->‘,res)

 

 

apply_async异步提交

def task(num):
    time.sleep(1)
    print(‘%s : %s‘%(num,os.getpid()))
    return num**2

if __name__ == ‘__main__‘:
    p = Pool()
    res_lst = [] #先添加进列表,最后再获取返回值
    for i in range(20):
        res = p.apply_async(task,args=(i,))   # 提交任务的方法 异步提交
        res_lst.append(res)
    for res in res_lst:
        print(res.get())

 

 

 

 

map方法

p = Pool() # 创建进程池

p.map(main,range(10))

# 将数组中的每个元素提取出来当作函数的参数，创建一个个进程，放进进程池中 # 第一个参数是函数，第二个参数是一个迭代器，将迭代器中的数字作为参数依次传入函数中

map方法是异步提交的简化版本,自带close和join方法,

返回值为一个列表

def task(num):
    time.sleep(0.3)
    print(‘%s : %s‘%(num,os.getpid()))
    return num**2
if __name__ == ‘__main__‘:
    p = Pool()
    ret=p.map(task,(range(20)))
    for i in ret:
        print(i)