Python进程池
Python进程池详细操作教程
可以使用与创建和使用线程池相同的方式来创建和使用进程池。流程池可以定义为一组预实例化和空闲的流程，它们随时可以进行工作。当我们需要执行大量任务时，与为每个任务实例化新流程相比，创建进程池更为可取。
Python模块– Concurrent.futures
Python标准库具有一个名为
concurrent.futures 的模块。在Python 3.2中添加了此模块，以为开发人员提供启动异步任务的高级接口。这是Python线程和多处理模块之上的抽象层，用于提供使用线程或进程池运行任务的接口。
在我们的后续部分中，我们将研究current.futures模块的不同子类。
Executor类
Executor是
concurrent.futures Python模块的抽象类。它不能直接使用，我们需要使用以下具体子类之一-
ThreadPoolExecutor
ProcessPoolExecutor
ProcessPoolExecutor –一个具体的子类
它是Executor类的具体子类之一。它使用了多重处理，我们得到了提交任务的流程池。该池将任务分配给可用进程，并安排它们运行。
如何创建ProcessPoolExecutor？
借助于
concurrent.futures 模块及其具体的子类
Executor ，我们可以轻松地创建一个过程池。为此，我们需要构造一个
ProcessPoolExecutor ，其中包含池中所需的进程数。默认情况下，该数字为5。然后将任务提交到进程池。
示例
我们现在将考虑创建线程池时使用的相同示例，唯一的区别是现在我们将使用
ProcessPoolExecutor 而不是
ThreadPoolExecutor 。
# Filename : example.py
# Copyright : 2020 By Lidihuo
# Author by : www.lidihuo.com
# Date : 2020-08-22
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()
输出
# Filename : example.py
# Copyright : 2020 By Lidihuo
# Author by : www.lidihuo.com
# Date : 2020-08-22
False
False
Completed
在上面的示例中，已使用5个线程构造了Process
PoolExecutor 。然后，将在发出消息之前等待2秒钟的任务被提交给进程池执行程序。从输出中可以看到，该任务要等到2秒后才能完成，因此第一次调用
done()将返回False。 2秒后，任务完成，我们通过调用
result()方法获得未来的结果。
实例化ProcessPoolExecutor –上下文管理器
实例化ProcessPoolExecutor的另一种方法是在上下文管理器的帮助下。它的工作原理与以上示例中使用的方法类似。使用上下文管理器的主要优点是它在语法上看起来不错。可以在以下代码的帮助下完成实例化-
# Filename : example.py
# Copyright : 2020 By Lidihuo
# Author by : www.lidihuo.com
# Date : 2020-08-22
with ProcessPoolExecutor(max_workers = 5) as executor
示例
为了更好地理解，我们采用与创建线程池时相同的示例。在此示例中，我们需要从导入
concurrent.futures 模块开始。然后创建一个名为
load_url()的函数，该函数将加载请求的URL。然后使用池中5个线程创建
ProcessPoolExecutor 。 Process
PoolExecutor 已被用作上下文管理器。我们可以通过调用
result()方法获得未来的结果。
# Filename : example.py
# Copyright : 2020 By Lidihuo
# Author by : www.lidihuo.com
# Date : 2020-08-22
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脚本将生成以下输出-
# Filename : example.py
# Copyright : 2020 By Lidihuo
# Author by : www.lidihuo.com
# Date : 2020-08-22
'http://some-made-up-domain.com/' generated an exception: <urlopen error [Errno 11004] getaddrinfo failed>
'http://www.foxnews.com/' page is 229476 bytes
'http://www.cnn.com/' page is 165323 bytes
'http://www.bbc.co.uk/' page is 284981 bytes
'http://europe.wsj.com/' page is 967575 bytes
使用Executor.map()函数
Python
map()函数被广泛用于执行许多任务。这样的任务之一是将特定功能应用于可迭代对象中的每个元素。同样，我们可以将迭代器的所有元素映射到一个函数，并将它们作为独立的作业提交给
ProcessPoolExecutor 。考虑下面的Python脚本示例以了解这一点。
示例
我们将考虑使用
Executor.map()函数创建线程池时使用的相同示例。在下面给出的示例中，map函数用于将
square()函数应用于values数组中的每个值。
# Filename : example.py
# Copyright : 2020 By Lidihuo
# Author by : www.lidihuo.com
# Date : 2020-08-22
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脚本将生成以下输出
# Filename : example.py
# Copyright : 2020 By Lidihuo
# Author by : www.lidihuo.com
# Date : 2020-08-22
4
9
16
25
何时使用ProcessPoolExecutor和ThreadPoolExecutor？
现在，我们已经研究了两个执行器类ThreadPoolExecutor和ProcessPoolExecutor，我们需要知道何时使用哪个执行器。对于受CPU约束的工作负载，我们需要选择ProcessPoolExecutor，而受I / O约束的工作负载，我们需要选择ThreadPoolExecutor。
如果我们使用
ProcessPoolExecutor ，那么我们就不必担心GIL，因为它使用了多处理。此外，与
ThreadPoolExecution 相比，执行时间会更少。考虑下面的Python脚本示例以了解这一点。
示例
# Filename : example.py
# Copyright : 2020 By Lidihuo
# Author by : www.lidihuo.com
# Date : 2020-08-22
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()
输出
# Filename : example.py
# Copyright : 2020 By Lidihuo
# Author by : www.lidihuo.com
# Date : 2020-08-22
Start: 8000000 Time taken: 1.5509998798370361
Start: 7000000 Time taken: 1.3259999752044678
Total time taken: 2.0840001106262207
Example- Python script with ThreadPoolExecutor:
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()
输出
# Filename : example.py
# Copyright : 2020 By Lidihuo
# Author by : www.lidihuo.com
# Date : 2020-08-22
Start: 8000000 Time taken: 3.8420000076293945
Start: 7000000 Time taken: 3.6010000705718994
Total time taken: 3.8480000495910645
从以上两个程序的输出中，我们可以看到使用
ProcessPoolExecutor 和
ThreadPoolExecutor 时执行时间的差异。
