二十五岁时我们都一样愚蠢、多愁善感,喜欢故弄玄虚,可如果不那样的话,五十岁时也就不会如此明智。
标题: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: <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时 执行时间的差异。