Python中的并发——线程池

假设我们需要为多线程任务创建大量线程。由于线程过多可能导致许多性能问题，这在计算上将非常昂贵，主要问题可能是吞吐量受限。我们可以通过创建线程池来解决这个问题。线程池可以定义为一组预实例化且处于空闲状态的线程，它们随时准备接收工作。当我们需要执行大量任务时，创建线程池优于为每个任务实例化新线程。线程池可以如下管理大量线程的并发执行：

如果线程池中的一个线程完成执行，则可以重用该线程。
如果一个线程终止，将创建另一个线程来替换该线程。

Python模块 – concurrent.futures

Python标准库包含**concurrent.futures**模块。此模块添加到Python 3.2中，为开发人员提供了一个高级接口来启动异步任务。它是Python的线程和多进程模块顶层的一个抽象层，用于提供使用线程池或进程池运行任务的接口。

在接下来的章节中，我们将学习concurrent.futures模块的不同类。

Executor类

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

ThreadPoolExecutor
ProcessPoolExecutor

ThreadPoolExecutor – 一个具体子类

它是Executor类的具体子类之一。该子类使用多线程，我们得到一个用于提交任务的线程池。该池将任务分配给可用的线程并安排它们运行。

如何创建一个ThreadPoolExecutor？

借助**concurrent.futures**模块及其具体子类**Executor**，我们可以轻松创建线程池。为此，我们需要构造一个**ThreadPoolExecutor**，其中包含我们想要在池中使用的线程数。默认情况下，数量为5。然后我们可以将任务提交到线程池。当我们**submit()**一个任务时，我们会得到一个**Future**。Future对象有一个名为**done()**的方法，它告诉我们future是否已解析。有了这个，该特定future对象的值已被设置。当任务完成时，线程池执行器将值设置为future对象。

示例

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

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

输出

False
True
Completed

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

实例化ThreadPoolExecutor – 上下文管理器

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

with ThreadPoolExecutor(max_workers = 5) as executor

示例

下面的示例取自Python文档。在这个例子中，首先必须导入**concurrent.futures**模块。然后创建一个名为**load_url()**的函数，该函数将加载请求的url。然后，该函数创建具有池中5个线程的**ThreadPoolExecutor**。**ThreadPoolExecutor**已被用作上下文管理器。我们可以通过在其上调用**result()**方法来获取future的结果。

import concurrent.futures
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()

with concurrent.futures.ThreadPoolExecutor(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)))

输出

以下是上述Python脚本的输出：

'http://some-made-up-domain.com/' generated an exception: <urlopen error [Errno 11004] getaddrinfo failed>
'http://www.foxnews.com/' page is 229313 bytes
'http://www.cnn.com/' page is 168933 bytes
'http://www.bbc.co.uk/' page is 283893 bytes
'http://europe.wsj.com/' page is 938109 bytes

Executor.map()函数的使用

Python **map()**函数广泛用于许多任务中。一项这样的任务是将某个函数应用于可迭代对象中的每个元素。类似地，我们可以将迭代器的所有元素映射到一个函数，并将这些元素作为独立作业提交到我们的**ThreadPoolExecutor**。考虑以下Python脚本示例，以了解该函数的工作方式。

示例

在下面的示例中，map函数用于将**square()**函数应用于values数组中的每个值。

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

输出

上述Python脚本生成以下输出：

打印页面