Concurrency In Python 简明教程

Concurrency in Python - Pool of Threads

假设我们必须为多线程任务创建一个大量线程。由于线程过多，它在计算上可能开销最大，因为可能出现很多性能问题。一个主要问题可能是吞吐量受限。我们可以通过创建一个线程池来解决此问题。线程池可以被定义为预实例化和空闲线程的组，它们随时准备接受工作。当我们需要完成大量任务时，相对于为每个任务实例化新线程，更赞成创建线程池。线程池可以按如下方式管理大量线程的并发执行 −

  1. 如果线程池中的线程完成其执行，那么该线程可以被重用。

  2. 如果一个线程终止，则会创建一个新线程来取代该线程。

Python Module – Concurrent.futures

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

在我们的后续部分，我们将了解 concurrent.futures 模块的不同类。

Executor Class

Executor*is an abstract class of the *concurrent.futures Python 模块。它不能直接使用，我们需要使用以下具体子类之一 −

  1. ThreadPoolExecutor

  2. ProcessPoolExecutor

ThreadPoolExecutor – A Concrete Subclass

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

How to create a ThreadPoolExecutor?

借助 concurrent.futures 模块及其具体子类 Executor ，我们可以很容易地创建一个线程池。为此，我们需要使用我们希望在池中实现的线程数来构建 ThreadPoolExecutor 。默认情况下，该数字为 5。然后，我们可以向线程池提交任务。当我们 submit() 一个任务，我们得到一个 Future 。将来对象有一个名为 done() 的方法，它告诉未来是否已解析。这样，已为该特定将来对象设置一个值。当一个任务完成，线程池执行程序会将值设置到将来对象。

Example

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

Output

False
True
Completed

在上面的示例中，已使用 5 个线程构建了一个 ThreadPoolExecutor 。然后，执行一个任务，在给出消息之前它将等待 2 秒，然后将此任务提交到线程池执行程序。从输出中可以看到，该任务执行 2 秒才完成，所以第一次调用的 done() 将返回 False。2 秒后，任务执行完毕，我们通过调用其 result() 方法来获取未来的结果。

Instantiating ThreadPoolExecutor – Context Manager

实例化 ThreadPoolExecutor 的另一种方法是借助上下文管理器。它的工作原理与上述示例中使用的该方法类似。主要使用上下文管理器的好处是它的语义很好。实例化可以通过以下代码来实现 −

with ThreadPoolExecutor(max_workers = 5) as executor

Example

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

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

Output

以下是上述 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

Use of Executor.map() function

Python map() 函数在许多任务中得到广泛使用。这类任务的一种是将某个函数应用于迭代中的每个元素。类似地，我们可以将迭代器的所有元素映射到一个函数，并将这些元素作为独立作业提交给 out ThreadPoolExecutor 。请看以下 Python 脚本示例，以了解函数是如何工作的。

Example

在下面的示例中，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()

Output

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

4
9
16
25