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

输出

False
False
Completed

在上面的示例中，已构造了一个具有5个线程的ProcessPoolExecutor。然后，将一个任务提交到进程池执行器，该任务将在发出消息之前等待2秒。从输出中可以看出，任务不会在2秒钟之前完成，因此对done()的第一次调用将返回False。2秒后，任务完成，我们通过对它调用result()方法来获取期货的结果。

实例化ProcessPoolExecutor – 上下文管理器

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

with ProcessPoolExecutor(max_workers = 5) as executor

示例

为了更好地理解，我们采用了与创建线程池时相同的示例。在此示例中，我们需要首先导入concurrent.futures模块。然后创建一个名为load_url()的函数，该函数将加载请求的URL。然后使用池中的5个线程数创建ProcessPoolExecutor。ProcessPoolExecutor已用作上下文管理器。我们可以通过对它调用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: <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数组中的每个值。

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

4
9
16
25

何时使用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时的执行时间差异。

打印页面
上一页
下一页
广告