python-进程池Pool

作者：马育民 • • 阅读：10356

概述

创建子进程数量不多时，可以使用Process，但创建 较多进程 时，用multiprocessing.Pool实现更为 简单

进程池

池

Pool是池的意思，为了讲解进程池，我们先说水池

特点：（类似小学、初中的数学题，某水池，一头进水，一头出水。。。）

  1. 水池有 最大容量
  2. 刚开始时，水池空，进水口进水，一直到 水池满水，不再进水，即使进水口有水，也不进水
  3. 当需要 大量用水 时，即 出水口大于进水口，如果没有水池，将供不应求。有水池后，在一段时间内，可以 提供大量用水
  4. 当水池的水不满时，如果 进水口有水 ，就 继续进水

进程池

特点：

  1. 初始化Pool时，可以指定最大进程数
  2. 如果进程池 没有满，就会 创建一个新的进程 执行任务
  3. 如果进程池已满，不再创建 进程，如果有 未执行的任务 则处于 等待
  4. 当某子进程执行任务结束，会 空闲 出来，不会退出，继续执行后面的任务，直到全部执行结束

注意：
进程池中的子进程全部启动后，不再 创建 新的子进程，而是利用 空闲的子进程 继续 执行 后面的 任务，这样可以 提高执行效率，因为创建进程需要 消耗系统资源

Pool的api介绍

导入模块

from multiprocessing import Pool

初始化进程池

Pool(n)

参数：

  • n：最大进程数，可以为空

apply() 几乎不用

添加任务，是 阻塞 的且支持结果返回。
注意：
此方法较少使用，一个任务执行后，才会执行下一个任务，而不是多进程并发执行，没有体现多进程的意义，所以较少使用

语法：

apply(func[, args=()[, kwds={}]])

apply_async()

添加任务，是非阻塞的且支持结果返回（需要调用get()方法），支持进行回调。

注意：
必须调用join()后才会启动进程

语法：

apply_async(func[, args=()[, kwds={}[, callback=None]]])

参数说明：

  • func：调用对象
  • args：表示调用对象的位置参数元组
  • kwargs：表示调用对象的关键字参数字典

map() 较少使用

注意：
此方法较少使用，一个任务执行后，才会执行下一个任务，而不是多进程并发执行，没有体现多进程的意义，所以较少使用

语法：

map(func, iterable[, chunksize=None])

Pool类中的map方法，与内置的map函数用法行为基本一致，它会使进程 阻塞 直到结果返回。
注意：虽然第二个参数是一个迭代器，但在实际使用中，必须在整个队列都就绪后，程序才会运行子进程。

map_async()

语法：

map_async(func, iterable[, chunksize[, callback]])

与map用法一致，但是它是非阻塞的。其有关事项见apply_async。

close()

关闭进程池，不在接受新的任务。

terminal()

结束工作进程，不在处理未处理的任务。

join()

主进程阻塞，等待子进程运行结束后，主进程才继续运行

通过apply_async()方法添加任务，必须调用join()方法才能 启动进程池

注意：
必须放在close()或terminate()之后使用，否则抛 “AssertionError 异常”

get()

获取调用对象的返回值
注意：
在循环添加任务时，不要调用get()方法，否则程序会 串行

Pool的使用

例子

4人的工程队修1000米公路，每人每次修路100米，通过进程池模拟实现

# coding=utf-8

from multiprocessing import Pool
import time,os

def task(num):
    print('%s号工人修100米，pid:%s'%(num,os.getpid()))
    time.sleep(3)
    print('%s号工人修100米完成'%num)


if __name__ == "__main__":
    print('4人的工程队修1000米公路')
    #初始化进程池，最大进程数时4
    p=Pool(4)
    #修1000米公路
    num=1
    length=1000
    while length!=0:
        #添加任务，第一次可以添加4个任务，之后有空闲进程时，才会继续添加任务
        p.apply_async(task,args=(num,))
        length-=100
        if num==4:
            num=1
        else:
            num+=1
    #关闭进程池，不在接受新的任务
    p.close()
    #主进程阻塞，启动进程池，只会启动4个进程，最多同时执行4个任务，当任务执行结束后，有进程空闲下来，再用空闲的进程执行下一个任务
    p.join()
    print('修1000米公路完成')

说明：
初始化进程池时，虽然数量时4，但可以添加所有任务，注意，只会启动4个进程 执行全部任务，当某进程执行结束后，空闲下来，再用该空闲的进程执行下一个任务。

注意：
进程池中的子进程全部启动后，不再 创建 新的子进程，而是利用 空闲的子进程 继续 执行 后面的 任务，这样可以 提高执行效率，因为创建进程需要 消耗系统资源

调用对象的返回值

有时执行任务后，要得到返回值，使用Pool可以很方便的实现

#coding=utf-8

from multiprocessing import Pool
import time,os,random

def xiulu(num):
    print('%s号工人开始修路,pid=%s'%(num,os.getpid()))
    #随机生成3~6的浮点数
    t=random.randint(3,6)
    #模拟修路的时长
    time.sleep(t)
    print('%s号工人修路100米完成,耗时：%s,pid=%s'%(num,t,os.getpid()))
    return t

if __name__ == "__main__":

    print('开始修路')

    p=Pool(4)
    num=1
    length=1000
    results=[]
    while length!=0:
        result=p.apply_async(func=xiulu,args=(num,))
        #此处不能用print()直接打印结果，因为get()方法会导致串行
        # print(result.get())
        results.append(result)
        if num==4:
            num=1
        else:
            num+=1
        length-=100

    p.close()
    p.join()
    for item in results:
        print(item.get())
    print('修路1000米完成')

例子（废弃）

#coding=utf-8

from multiprocessing import Pool
import os
import time

#假设该函数是耗时代码
def task(*args):
    print('子进程-%d 开始,pid=%d,ppid=%d'%(args[0],os.getpid(),os.getppid()))
    time.sleep(args[0])

if __name__ == "__main__":
    print('主进程pid=%d'%os.getpid())

    #初始化进程池，最大进程数是3
    p=Pool(3)
    for index in range(8):
        #添加任务，虽然进程池最大进程数时3，但可以添加所有任务
        p.apply_async(task,args=(index,))
    #关闭进程池，不在接受新的任务
    p.close()
    #主进程阻塞，启动进程池，只会启动3个进程执行3个任务，当任务执行后，有进程空闲下来，再用空闲的进程执行下一个任务
    p.join()
    print('主进程结束')

说明：
初始化进程池时，虽然数量时3，但可以添加所有任务，注意，只会启动3个进程 执行3个任务，当任务执行后，有进程空闲下来，再用空闲的进程执行下一个任务
注意：
没有再次创建进程

执行结果：

任务截图：

类比：

相当于饮水机，只有两个水龙头，可以给n个杯子灌水，但同时只给2个杯子灌水，灌满后，再用空闲的水龙头给下一个杯子灌水，而不是再添加一个水龙头

调用对象的返回值（废弃）

有时执行任务后，要得到返回值，使用Pool可以很方便的实现

#coding=utf-8

from multiprocessing import Pool
import os
import time

#假设该函数是耗时代码
def task(*args):
    print('子进程-%d 开始,pid=%d,ppid=%d'%(args[0],os.getpid(),os.getppid()))
    num=0
    for item in range(30000000):
        num+=1
    return num

if __name__ == "__main__":
    starttime=time.time()
    print('主进程pid=%d'%os.getpid())
    #定义list，存放返回的结果
    results=[]
    #初始化进程池，最大进程数时3
    p=Pool(3)
    for index in range(8):
        #添加任务，可以添加所有任务
        r=p.apply_async(task,(index,))
        results.append(r)
        #如果直接打印结果，虽然创建多个进程，但其实是一个进程运行结束后，才运行下一个进程
        # print(r.get())
    #关闭进程池，不在接受新的任务
    p.close()
    #主进程阻塞，启动进程，只会启动3个进程执行3个任务，当任务执行后，有进程空闲下来，再用空闲的进程执行下一个任务
    p.join()
    print('主进程结束')

    for item in results:
        print(item.get())

    endtime=time.time()
    print('耗时：%d'%(endtime-starttime))

进程池的初始数量

进程池的初始数量，与硬件有关，如cpu数量，cpu核心数量，内存大小等，所以具体设置成多少，需要具体情况具体分析，并且需要反复测试运行效率，最终得到合理的数量

原文出处：/show_1EF2o4y2hFxV.html

吉ICP备15003043号