假设我们必须为多线程任务创建大量线程。由于线程太多,因此可能存在许多性能问题,这在计算上是最昂贵的。一个主要问题可能是吞吐量受限。我们可以通过创建一个线程池来解决这个问题。线程池可以被定义为预先实例化和空闲线程的组,其准备好被给予工作。当我们需要执行大量任务时,创建线程池优先于为每个任务实例化新线程。线程池可以管理大量线程的并发执行,如下所示
- 如果线程池中的线程完成其执行,则可以重用该线程。
- 如果线程终止,则将创建另一个线程来替换该线程。
python模块 - concurrent.futures
python标准库包括 concurrent.futures 模块。该模块是在python 3.2中添加的,用于为开发人员提供启动异步任务的高级接口。它是python的线程和多处理模块之上的抽象层,用于提供使用线程池或进程池运行任务的接口。
在接下来的部分中,我们将了解concurrent.futures模块的不同类。
执行者类
executor 是 concurrent.futures python模块 的抽象类。它不能直接使用,我们需要使用以下具体子类之一
- 的threadpoolexecutor
- processpoolexecutor
threadpoolexecutor - 一个具体的子类
它是executor类的具体子类之一。子类使用多线程,我们获得了一个用于提交任务的线程池。此池将任务分配给可用线程并安排它们运行。
如何创建threadpoolexecutor?
在 concurrent.futures 模块及其具体的子类 executor 的帮助下,我们可以轻松地创建一个线程池。为此,我们需要构造一个 threadpoolexecutor ,其中包含我们在池中想要的线程数。默认情况下,该数字为5.然后我们可以向线程池提交任务。当我们 提交() 任务时,我们会回到 未来。future对象有一个名为 done() 的方法,它告诉未来是否已经解决。有了这个,就为该特定的未来对象设置了一个值。任务完成后,线程池执行程序将值设置为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
在上面的例子中, threadpoolexecutor 已经构造了5个线程。然后,在给出消息之前等待2秒的任务被提交给线程池执行器。从输出中可以看出,任务直到2秒才完成,因此第一次调用 done() 将返回false。2秒后,任务完成,我们通过调用 result() 方法得到未来的 结果 。
实例化threadpoolexecutor - 上下文管理器
实例化 threadpoolexecutor的 另一种方法是在上下文管理器的帮助下。它的工作方式与上例中使用的方法类似。使用上下文管理器的主要优点是它在语法上看起来很好。实例化可以在以下代码的帮助下完成
with threadpoolexecutor(max_workers = 5) as executor
例
以下示例是从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)))
输出
以下是上述python脚本的输出
'http://some-made-up-domain.com/' generated an exception: '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脚本生成以下输出 -
4 9 16 25