进程池Pool

在利用Python进行系统管理的时候，特别是同时操作多个文件目录，或者远程控制多台主机，并行操作可以节约大量的时间。当被操作对象数目不大时，可以直接利用multiprocessing中的Process动态成生多个进程，但如果是上百个，上千个目标，手动的去限制进程数量却又太过繁琐，此时就可以用到multiprocessing模块提供的Pool方法。

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

一：使用进程池

例1：非阻塞

from multiprocessing import Pool

import os, time, random

def worker(name):

t_start = time.time()

print("%s开始执行,进程号为%d" % (name, os.getpid()))

# random.random()随机生成0~1之间的浮点数

time.sleep(random.random() * 2)

t_stop = time.time()

print(name, "执行完毕，耗时%0.2f" % (t_stop - t_start))

def main():

po = Pool(5) # 定义一个进程池，最大进程数5

# 往进程池中添加任务

for i in range(10):

# Pool.apply_async(要调用的目标,(传递给目标的参数元祖,))

# 每次循环将会用空闲出来的子进程去调用目标

po.apply_async(worker, (f'liang{i}',))

print("----start----")

po.close() # 关闭进程池，关闭后po不再接收新的请求

po.join() # 等待po中所有子进程执行完成，必须放在close语句之后

print("----all_done----")

if __name__ == '__main__':

main()

执行结果

----start----

liang0开始执行,进程号为10404

liang1开始执行,进程号为9920

liang2开始执行,进程号为13136

liang3开始执行,进程号为10180

liang4开始执行,进程号为7708

liang4 执行完毕，耗时0.57

liang5开始执行,进程号为7708

liang2 执行完毕，耗时1.20

liang6开始执行,进程号为13136

liang1 执行完毕，耗时1.33

liang7开始执行,进程号为9920

liang0 执行完毕，耗时1.34

liang8开始执行,进程号为10404

liang3 执行完毕，耗时1.96

liang9开始执行,进程号为10180

liang5 执行完毕，耗时1.73

liang9 执行完毕，耗时0.54

liang8 执行完毕，耗时1.28

liang7 执行完毕，耗时1.37

liang6 执行完毕，耗时1.88

----all_done----

函数解释：

apply_async(func[, args[, kwds]]) ：使用非阻塞方式调用func(并行执行，堵塞方式必须等待上一个进程退出才能执行下一个进程)，args为传递给func的参数列表, kwds为传递给func的关键字参数列表；

apply(func[, args[, kwds]])：使用阻塞方式调用func

close()：关闭Pool，使其不再接受新的任务；

terminate()：不管任务是否完成，立即终止；

join()：主进程阻塞，等待子进程的退出， 必须在close或terminate之后使用；

执行说明：创建一个进程池pool，并设定进程的数量为5，range(10)会相继产生10个对象，10个对象被提交到pool中，因pool指定进程数为5，所以0、1、2、3、4会直接送到进程中执行，当其中一个执行完后才空出一个进程处理对象，继续去执行新的对象，所以会出现输出“liang5开始执行,进程号为7708”出现在"liang4 执行完毕，耗时0.57"后。因为为非阻塞，主函数会自己执行自个的，不搭理进程的执行，所以运行完for循环后直接输出“----start----”，主程序在pool.join()处等待各个进程的结束。

例2：阻塞

from multiprocessing import Pool

import os, time, random

def worker(name):

t_start = time.time()

print("%s开始执行,进程号为%d" % (name, os.getpid()))

# random.random()随机生成0~1之间的浮点数

time.sleep(random.random() * 2)

t_stop = time.time()

print(name, "执行完毕，耗时%0.2f" % (t_stop - t_start))

def main():

po = Pool(3) # 定义一个进程池，最大进程数3

# 往进程池中添加任务

for i in range(0, 5):

# Pool.apply_async(要调用的目标,(传递给目标的参数元祖,))

# 每次循环将会用空闲出来的子进程去调用目标

po.apply(worker, (f'liang{i}',))

print("----start----")

po.close() # 关闭进程池，关闭后po不再接收新的请求

po.join() # 等待po中所有子进程执行完成，必须放在close语句之后

print("----all_done----")

if __name__ == '__main__':

main()

输出

liang0开始执行,进程号为1976

liang0 执行完毕，耗时1.75

liang1开始执行,进程号为12624

liang1 执行完毕，耗时0.57

liang2开始执行,进程号为12444

liang2 执行完毕，耗时0.52

liang3开始执行,进程号为1976

liang3 执行完毕，耗时1.23

liang4开始执行,进程号为12624

liang4 执行完毕，耗时0.85

----start----

----all_done----

因为是阻塞，主函数会等待进程的执行，执行完之后才会继续往下，所以运行完所有进程后才输出“----start----”

例3、使用进程池，并返回结果

from multiprocessing import Pool

import os, time, random

def worker(name):

print("%s开始执行,进程号为%d" % (name, os.getpid()))

# random.random()随机生成0~1之间的浮点数

time.sleep(random.random() * 2)

return name,os.getpid()

def main():

po = Pool(3) # 定义一个进程池，最大进程数3

res=[]

# 往进程池中添加任务

for i in range(0, 5):

# Pool.apply_async(要调用的目标,(传递给目标的参数元祖,))

# 每次循环将会用空闲出来的子进程去调用目标

res.append(po.apply_async(worker, (f'liang{i}',)))

print("----start----")

po.close() # 关闭进程池，关闭后po不再接收新的请求

po.join() # 等待po中所有子进程执行完成，必须放在close语句之后

for result in res:

print(result.get()) #get()函数得出每个返回结果的值

print("----all_done----")

if __name__ == '__main__':

main()

输出结果：

----start----

liang0开始执行,进程号为14012

liang1开始执行,进程号为13000

liang2开始执行,进程号为14120

liang3开始执行,进程号为14012

liang4开始执行,进程号为14012

('liang0', 14012)

('liang1', 13000)

('liang2', 14120)

('liang3', 14012)

('liang4', 14012)

----all_done----

例4、多进程执行多个任务

from multiprocessing import Pool

import os, time, random

def worker1(name):

print("%s开始执行work1,进程号为%d" % (name, os.getpid()))

# random.random()随机生成0~1之间的浮点数

time.sleep(random.random() * 2)

def worker2(name):

print("%s开始执行work2,进程号为%d" % (name, os.getpid()))

# random.random()随机生成0~1之间的浮点数

time.sleep(random.random() * 2)

def worker3(name):

print("%s开始执行work3,进程号为%d" % (name, os.getpid()))

# random.random()随机生成0~1之间的浮点数

time.sleep(random.random() * 2)

def main():

po = Pool(4) # 定义一个进程池，最大进程数3

work_list=[worker1,worker2,worker3]

# 往进程池中添加任务

for work in work_list:

for i in range(3):

po.apply_async(work, (f'liang{i}',))

print("----start----")

po.close() # 关闭进程池，关闭后po不再接收新的请求

po.join() # 等待po中所有子进程执行完成，必须放在close语句之后

print("----all_done----")

if __name__ == '__main__':

main()

线程池4个线程执行3个任务，每个任务执行3次。

输出：

----start----

liang0开始执行work1,进程号为13088

liang1开始执行work1,进程号为4908

liang2开始执行work1,进程号为4200

liang0开始执行work2,进程号为8124

liang1开始执行work2,进程号为4908

liang2开始执行work2,进程号为13088

liang0开始执行work3,进程号为8124

liang1开始执行work3,进程号为4200

liang2开始执行work3,进程号为4908

----all_done----

二、进程池进程之间的通讯

进程池中进程的通讯队列

from multiprocessing import Pool, Manager

q = Manager().Queue()

import os

import time

from multiprocessing import Pool, Manager

def work(name, q):

time.sleep(1)

print(f"{name}：---{os.getpid()}---{q.get()}")

def main():

# 创建一个用于进程池通信的队列

q = Manager().Queue()

for i in range(1000):

q.put(f'data-{i}')

# 创建一个拥有五个进程的进程池

po = Pool(5)

# 往进程池中添加20个任务

for i in range(20):

po.apply_async(work, (f'liang{i}', q))

# close:关闭进程池(进程池停止接收任务)

po.close()

# 主进程等待进程池中的任务结束再往下执行

po.join()

if __name__ == '__main__':

main()