Python 进程池

python 进程池

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

python模块 - concurrent.futures

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

在接下来的部分中，我们将查看concurrent.futures模块的不同子类。

执行者类

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()

输出

false
false
completed

在上面的示例中，process poolexecutor 已构造为5个线程。然后，在给出消息之前等待2秒的任务被提交给进程池执行器。从输出中可以看出，任务直到2秒才完成，因此第一次调用 done（）将返回false。2秒后，任务完成，我们通过调用 result（）方法得到未来的结果。

实例化processpoolexecutor - 上下文管理器

实例化processpoolexecutor的另一种方法是在上下文管理器的帮助下。它的工作方式与上例中使用的方法类似。使用上下文管理器的主要优点是它在语法上看起来很好。实例化可以在以下代码的帮助下完成

with processpoolexecutor(max_workers = 5) as executor

例

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

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脚本将生成以下输出

'http://some-made-up-domain.com/' generated an exception: 
'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数组中的每个值。

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，我们需要知道何时使用哪个执行器。我们需要在遇到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()

输出

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()

输出

start: 8000000 time taken: 3.8420000076293945
start: 7000000 time taken: 3.6010000705718994
total time taken: 3.8480000495910645

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

下一节：python 多处理器

Python技巧