Python - 并发性进程池

进程池的创建和使用方式与我们创建和使用线程池的方式相同。进程池可以定义为一组预先实例化的进程和空闲进程，它们随时准备被分配工作。当我们需要执行大量任务时，创建进程池比为每个任务实例化新进程更可取。

Python 模块 – Concurrent.futures

Python 标准库有一个名为 concurrent.futures 的模块。此模块是在 Python 3.2 中添加的，用于为开发人员提供用于启动异步任务的高级接口。它是 Python 的线程和多处理模块之上的抽象层，用于提供使用线程池或进程运行任务的接口。

在后续的章节中，我们将了解 concurrent.futures 模块的不同子类。

Executor 类

Executor 是 concurrent.futures Python 模块的抽象类。它不能直接使用，我们需要使用以下具体子类之一 -

• ThreadPoolExecutor
• ProcessPoolExecutor

ProcessPoolExecutor – 一个具体的子类

它是 Executor 类的具体子类之一。它使用多进程，我们得到了一个用于提交任务的进程池。此池将任务分配给可用进程并计划它们运行。

如何创建 ProcessPoolExecutor？

在 concurrent.futures 模块及其具体子类 Executor 的帮助下，我们可以轻松地创建一个流程池。为此，我们需要构造一个 ProcessPoolExecutor，其中包含我们想要的池中的进程数。默认情况下，该数字为 5。然后，将任务提交到进程池。

现在，我们将考虑在创建线程池时使用的相同示例，唯一的区别是现在我们将使用 ProcessPoolExecutor 而不是 ThreadPoolExecutor 。

from concurrent.futures import ProcessPoolExecutor
from time import sleep
def task(message):
	 	sleep(2)
	 	return message

def main():
	 	executor = ProcessPoolExecutor(5)
	 	future = executor.submit(task, ("Completed"))
	 	print(future.done())
	 	sleep(2)
	 	print(future.done())
	 	print(future.result())
if __name__ == '__main__':
main()

输出

在上面的示例中，构造了一个具有 5 个线程的 ProcessPoolExecutor。然后，一个任务（在给出消息之前将等待 2 秒）提交到进程池执行程序。从输出中可以看出，任务直到 2 秒才完成，因此第一次调用 done（） 将返回 False。2 秒后，任务完成，我们通过对其调用 result（） 方法来获得 future 的结果。

实例化 ProcessPoolExecutor – 上下文管理器

实例化 ProcessPoolExecutor 的另一种方法是借助上下文管理器。它的工作原理类似于上面示例中使用的方法。使用上下文管理器的主要优点是它在语法上看起来不错。实例化可以借助以下代码完成 -

with ProcessPoolExecutor(max_workers = 5) as executor

例

为了更好地理解，我们采用与创建线程池时相同的示例。在此示例中，我们需要从导入 concurrent.futures 模块开始。然后创建一个名为 load_url（） 的函数，该函数将加载请求的 url。然后，使用池中的 5 个线程创建 ProcessPoolExecutor。ProcessPoolExecutor 已用作上下文管理器。我们可以通过对它调用 result（） 方法来获得 future 的结果。

import concurrent.futures
from concurrent.futures import ProcessPoolExecutor
import urllib.request

URLS = ['http://www.foxnews.com/',
	 	'http://www.cnn.com/',
	 	'http://europe.wsj.com/',
	 	'http://www.bbc.co.uk/',
	 	'http://some-made-up-domain.com/']

def load_url(url, timeout):
	 	with urllib.request.urlopen(url, timeout = timeout) as conn:
	 	 	 return conn.read()

def main():
	 	with concurrent.futures.ProcessPoolExecutor(max_workers=5) as executor:
	 	 	 future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
	 	 	 for future in concurrent.futures.as_completed(future_to_url):
	 	 	 url = future_to_url[future]
	 	 	 try:
	 	 	 	 	data = future.result()
	 	 	 except Exception as exc:
	 	 	 	 	print('%r generated an exception: %s' % (url, exc))
	 	 	 else:
	 	 	 	 	print('%r page is %d bytes' % (url, len(data)))

if __name__ == '__main__':
	 	main()

输出

上面的 Python 脚本将生成以下输出 -

使用 Executor.map（） 函数

Python map（） 函数广泛用于执行许多任务。其中一项任务是将某个函数应用于可迭代对象中的每个元素。同样，我们可以将迭代器的所有元素映射到一个函数，并将这些元素作为独立的作业提交给 ProcessPoolExecutor。请考虑以下 Python 脚本示例来理解这一点。

我们将考虑使用 Executor.map（） 函数创建线程池时使用的相同示例。在下面给出的示例中，map 函数用于将 square（） 函数应用于 values 数组中的每个值。

from concurrent.futures import ProcessPoolExecutor
from concurrent.futures import as_completed
values = [2,3,4,5]
def square(n):
	 	return n * n
def main():
	 	with ProcessPoolExecutor(max_workers = 3) as executor:
	 	 	 results = executor.map(square, values)
	 	for result in results:
	 	 	 print(result)
if __name__ == '__main__':
	 	main()

输出

上面的 Python 脚本将生成以下输出

何时使用 ProcessPoolExecutor 和 ThreadPoolExecutor？

现在我们已经研究了 Executor 类 —— ThreadPoolExecutor 和 ProcessPoolExecutor，我们需要知道何时使用哪个 executor。如果是 CPU 密集型工作负载，我们需要选择 ProcessPoolExecutor，如果是 I/O 密集型工作负载，我们需要选择 ThreadPoolExecutor。

如果我们使用 ProcessPoolExecutor，那么我们就不需要担心 GIL，因为它使用了多处理。此外，与 ThreadPoolExecution 相比，执行时间会更短。请考虑以下 Python 脚本示例来理解这一点。

例

import time
import concurrent.futures

value = [8000000, 7000000]

def counting(n):
	 	start = time.time()
	 	while n > 0:
	 	 	 n -= 1
	 	return time.time() - start

def main():
	 	start = time.time()
	 	with concurrent.futures.ProcessPoolExecutor() as executor:
	 	 	 for number, time_taken in zip(value, executor.map(counting, value)):
	 	 	 	 	print('Start: {} Time taken: {}'.format(number, time_taken))
	 	print('Total time taken: {}'.format(time.time() - start))

if __name__ == '__main__':
main()

输出

示例-带有ThreadPoolExecutor的Python脚本：

import time
import concurrent.futures

value = [8000000, 7000000]

def counting(n):
			start = time.time()
			while n > 0:
						n -= 1
			return time.time() - start

def main():
			start = time.time()
			with concurrent.futures.ThreadPoolExecutor() as executor:
						for number, time_taken in zip(value, executor.map(counting, value)):
									print('Start: {} Time taken: {}'.format(number, time_taken))
						print('Total time taken: {}'.format(time.time() - start))

if __name__ == '__main__':
main()

输出

从上面两个程序的输出中，我们可以看到使用 ProcessPoolExecutor 和 ThreadPoolExecutor 时执行时间的差异。