From:https://www.cnblogs.com/weihengblog/p/9812110.html
concurrent.futures 官方文档:https://docs.python.org/3/library/concurrent.futures.html
concurrent.futures: 线程池, 让你更加高效, 并发的处理任务:https://www.h3399.cn/201906/703751.html
python 因为其全局解释器锁 GIL 而无法通过线程实现真正的平行计算。这个论断我们不展开,但是有个概念我们要说明,
IO 密集型 vs 计算密集型:
- IO密集型:读取文件,读取网络套接字频繁。
- 计算密集型:大量消耗CPU的数学与逻辑运算,也就是我们这里说的平行计算。
而 concurrent.futures 模块,可以利用 multiprocessing 实现真正的平行计算。
核心原理是:concurrent.futures 会以子进程的形式,平行的运行多个 python 解释器,从而令 python 程序可以利用多核 CPU 来提升执行速度。由于 子进程 与 主解释器 相分离,所以他们的全局解释器锁也是相互独立的。每个子进程都能够完整的使用一个CPU 内核。
Python 模块 - Concurrent.futures
从 Python3.2开始,Python 标准库提供了 concurrent.futures 模块,为开发人员提供了启动异步任务的高级接口。 它提供了 ThreadPoolExecutor 和 ProcessPoolExecutor 两个类,实现了对 threading 和 multiprocessing 的更高级的抽象,对编写 线程池/进程池 提供了直接的支持。 可以将相应的 tasks 直接放入线程池/进程池,不需要维护Queue来操心死锁的问题,线程池/进程池会自动帮我们调度。
Future总结
1. python3自带,python2需要安装 2. Executer对象 它是一个抽象类,它提供了异步执行的方法,他不能直接使用,但可以通过它的子类 ThreadPoolExecuter和ProcessPoolExecuter 2.1 Executer.submit(fn, *args, **kwargs) fn: 需要异步执行的函数 *args,**kwargs fn 接受的参数 该方法的作用就是提交一个可执行的回调 task,它返回一个 Future 对象 2.2 map(fn, *iterables, timeout=None, chunksize=1) map(task,URLS) # 返回一个 map()迭代器,这个迭代器中的回调执行返回的结果是有序的 3. Future对象相关 future可以理解为一个在未来完成的操作,这是异步编程的基础 通常情况下我们在遇到IO操作的时候,将会发生阻塞,cpu不能做其他事情 而future的引入帮助我们在这段等待时间可以完成其他的操作 3.1 done(): 如果当前线程已取消/已成功,返回True。 3.2 cance(): 如果当前线程正在执行,并且不能取消调用,返回Flase。否则调用取消,返回True 3.3 running(): 如果当前的线程正在执行,则返回True 3.4 result(): 返回调用返回的值,如果调用尚未完成,则此方法等待 如果等待超时,会抛出concurrent.futures.TimeoutError 如果没有指定超时时间,则等待无时间限制 如果在完成之前,取消了Future,则会引发CancelledError 4. as_completed(): 在多个Future实例上的迭代器将会被返回 这些Future实例由fs完成时产生。 由fs返回的任何重复的Future,都会被返回一次。 里面保存的都是已经执行完成的Future对象 5. wait(): 返回一个元祖,元祖包含两个元素 1. 已完成的future集合 2. 未完成的future集合
初体验:
-
# coding=utf-8 from concurrent import futures from concurrent.futures import Future import time def return_future(msg): time.sleep(3) return msg pool = futures.ThreadPoolExecutor(max_workers=2) t1 = pool.submit(return_future,'hello') t2 = pool.submit(return_future,'world') time.sleep(3) print(t1.done()) # 如果顺利完成,则返回True time.sleep(3) print(t2.done()) print(t1.result()) # 获取future的返回值 time.sleep(3) print(t2.result()) print("主线程")
map(func,* iterables,timeout = None,chunksize = 1 )
# coding=utf-8 import time from concurrent.futures import Future,as_completed from concurrent.futures import ThreadPoolExecutor as Pool import requests import time URLS = ['http://www.baidu.com', 'http://qq.com', 'http://sina.com'] def task(url,timeout=10): return requests.get(url=url,timeout=timeout) pool = Pool() result = pool.map(task,URLS) start_time = time.time() # 按照 URLS 的顺序返回 for res in result: print("{} {}".format(res.url,len(res.content))) # 无序的 with Pool(max_workers=3) as executer: future_task = [executer.submit(task,url) for url in URLS] for f in as_completed(future_task): if f.done(): f_ret = f.result() # f.result()得到task的返回值,requests对象 print('%s, done, result: %s, %s' % (str(f), f_ret.url, len(f_ret.content))) print("耗时", time.time() - start_time) print("主线程")
Future对象
Future可以理解为一个未来完成的操作
当我们执行io操作的时候,在等待返回结果之前会产生阻塞
cpu不能做其他事情,而Future的引入帮助我们在等待的这段时间可以完成其他操作
from concurrent.futures import ThreadPoolExecutor as Pool from concurrent.futures import as_completed import requests import time URLS = ['http://www.baidu.com', 'http://qq.com', 'http://sina.com'] def task(url,timeout=10): return requests.get(url=url,timeout=timeout) # start_time = time.time() # for url in URLS: # ret = task(url) # print("{} {}".format(ret.url,len(ret.content))) # print("耗时",time.time() - start_time) with Pool(max_workers=3) as executor: # 创建future任务 future_task = [executor.submit(task,url) for url in URLS] for f in future_task: if f.running(): print("%s is running"%str(f)) for f in as_completed(future_task): try: ret = f.done() if ret: f_ret = f.result() print('%s, done, result: %s, %s' % (str(f), f_ret.url, len(f_ret.content))) except Exception as e: f.cance() print(e) """ url不是按照顺序返回的,说明并发时,当访问某一个url时,如果没有得到返回结果,不会发生阻塞 <Future at 0x1c63990e6d8 state=running> is running <Future at 0x1c639922780 state=running> is running <Future at 0x1c639922d30 state=running> is running <Future at 0x1c63990e6d8 state=finished returned Response>, done, result: http://www.baidu.com/, 2381 <Future at 0x1c639922780 state=finished returned Response>, done, result: https://www.qq.com?fromdefault, 243101 <Future at 0x1c639922d30 state=finished returned Response>, done, result: http://sina.com/, 23103 """
模块方法
concurrent.futures.wait(fs, timeout=None, return_when=ALL_COMPLETED)
wait() 会返回一个tuple,tuple 会包含两个集合:已完成的集合 和 未完成的集合。使用 wait() 会获得更大的自由度,他接受三个参数:FIRST_COMPLETED、FIRST_EXCEPTION、ALL_COMPLETE。默认为 ALL_COMPLETE。
如果采用默认的 ALL_COMPLETED,程序会阻塞直到线程池里面的所有任务都完成,再执行主线程:
from concurrent.futures import Future from concurrent.futures import ThreadPoolExecutor as Pool from concurrent.futures import as_completed, wait import requests URLS = ['http://www.baidu.com', 'http://qq.com', 'http://sina.com'] def task(url, timeout=10): r = requests.get(url=url, timeout=timeout) print(r.status_code) with Pool(max_workers=3) as execute: future_task = [execute.submit(task, url) for url in URLS] for f in future_task: if f.running(): print("%s" % (str(f))) """ 并且wait还有timeout和return_when两个参数 return_when有三个常量 (默认是 ALL_COMPLETED) FIRST_COMPLETED 任何一个future_task执行完成时/取消时,改函数返回 FIRST_EXCEPTION 任何一个future_task发生异常时,该函数返回,如果没有异常发生,等同于ALL_COMPLETED ALL_COMPLETED 当所有的future_task执行完毕返回。 """ results = wait(future_task, return_when="FIRST_COMPLETED") # done = results[0] for d in done: print(d)
concurrent.futures.as_completed(fs, timeout=None)
在多个 Future 实例上的迭代器将会被返回,这些 Future 实例由 fs 完成时产生。由 fs 返回的任何重复的 Future,都会被返回一次。里面保存的都是已经执行完成的 Future 对象。
from concurrent.futures import ThreadPoolExecutor as Pool from concurrent.futures import as_completed import requests import time URLS = ['http://www.baidu.com', 'http://qq.com', 'http://sina.com'] def task(url,timeout=10): return requests.get(url=url,timeout=timeout) with Pool(max_workers=3) as executor: # 创建future任务 future_task = [executor.submit(task,url) for url in URLS] for f in future_task: if f.running(): print("%s is running"%str(f)) for f in as_completed(future_task): try: ret = f.done() if ret: f_ret = f.result() print('%s, done, result: %s, %s' % (str(f), f_ret.url, len(f_ret.content))) except Exception as e: f.cance() print(e)
下面我们将学习 concurrent.futures
模块中的类。concurrent.futures 基础模块是 executor 和 future。
使用示例代码:
# -*- coding:utf-8 -*- import redis from redis import WatchError from concurrent.futures import ProcessPoolExecutor r = redis.Redis(host='127.0.0.1', port=6379) # 减库存函数, 循环直到减库存完成 # 库存充足, 减库存成功, 返回True # 库存不足, 减库存失败, 返回False def reduce_stock(): # python中redis事务是通过pipeline的封装实现的 with r.pipeline() as pipe: while True: try: # watch库存键, multi后如果该key被其他客户端改变, 事务操作会抛出WatchError异常 pipe.watch('stock:count') count = int(pipe.get('stock:count')) if count > 0: # 有库存 # 事务开始 pipe.multi() pipe.decr('stock:count') # 把命令推送过去 # execute返回命令执行结果列表, 这里只有一个decr返回当前值 print(pipe.execute()[0]) return True else: return False except WatchError as ex: # 打印WatchError异常, 观察被watch锁住的情况 print(ex) pipe.unwatch() def worker(): while True: # 没有库存就退出 if not reduce_stock(): break if __name__ == "__main__": # 设置库存为100 r.set("stock:count", 100) # 多进程模拟多个客户端提交 with ProcessPoolExecutor() as pool: for _ in range(10): pool.submit(worker)
concurrent.futures 模块详解
1. Executor对象
class concurrent.futures.Executor
Executor是一个抽象类,它提供了异步执行调用的方法。它不能直接使用,但可以通过它的两个子类ThreadPoolExecutor或者ProcessPoolExecutor进行调用。
1.1 Executor.submit(fn, *args, **kwargs)
fn:需要异步执行的函数
*args, **kwargs:fn 的参数
示例代码:
# -*- coding:utf-8 -*- from concurrent import futures def test(num): import time return time.ctime(), num with futures.ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(test, 1) print(future.result())
1.2 Executor.map(func, *iterables, timeout=None)
相当于map(func, *iterables),但是func是异步执行。timeout的值可以是int或float,如果操作超时,会返回raisesTimeoutError;如果不指定timeout参数,则不设置超时间。
func:需要异步执行的函数
*iterables:可迭代对象,如列表等。每一次func执行,都会从iterables中取参数。
timeout:设置每次异步操作的超时时间
示例代码:
# -*- coding:utf-8 -*- from concurrent import futures def test(num): import time return time.ctime(), num data = [1, 2, 3] with futures.ThreadPoolExecutor(max_workers=1) as executor: for future in executor.map(test, data): print(future)
1.3 Executor.shutdown(wait=True)
释放系统资源,在Executor.submit()或 Executor.map()等异步操作后调用。使用with语句可以避免显式调用此方法。
2. ThreadPoolExecutor对象
ThreadPoolExecutor类是Executor子类,使用线程池执行异步调用.
class concurrent.futures.ThreadPoolExecutor(max_workers),使用 max_workers 数目的线程池执行异步调用
3. ProcessPoolExecutor对象
ThreadPoolExecutor类是Executor子类,使用进程池执行异步调用.
class concurrent.futures.ProcessPoolExecutor(max_workers=None),使用 max_workers数目的进程池执行异步调用,如果max_workers为None则使用机器的处理器数目(如4核机器max_worker配置为None时,则使用4个进程进行异步并发)。
示例代码:
# -*- coding:utf-8 -*- from concurrent import futures def test(num): import time return time.ctime(), num def muti_exec(m, n): # m 并发次数 # n 运行次数 with futures.ProcessPoolExecutor(max_workers=m) as executor: # 多进程 # with futures.ThreadPoolExecutor(max_workers=m) as executor: #多线程 executor_dict = dict((executor.submit(test, times), times) for times in range(m * n)) for future in futures.as_completed(executor_dict): times = executor_dict[future] if future.exception() is not None: print('%r generated an exception: %s' % (times, future.exception())) else: print('RunTimes:%d,Res:%s' % (times, future.result())) if __name__ == '__main__': muti_exec(5, 1)
调度单个任务
执行者类Executor调度单个任务,使用submit() 函数,然后用返回的 Future 实例等待任务结果。
Executor 是一个 Python concurrent.futures
模块的抽象类。 它不能直接使用,我们需要使用以下具体子类之一 -
ThreadPoolExecutor:线程池
ProcessPoolExecutor:进程池
示例代码:
from concurrent import futures import time import random def task(n): time.sleep(random.randint(1, 10)) return n executor = futures.ThreadPoolExecutor(max_workers=3) future = executor.submit(task, 5) print('future: {}'.format(future)) result = future.result() print('result: {}'.format(result))
线程池 和 进程池
- ThreadPoolExecutor 是
Executor
类的具体子类之一。 子类使用多线程,我们得到一个提交任务的线程池。 该池将任务分配给可用线程并安排它们运行。 - ProcessPoolExecutor 是
Executor
类的具体子类之一。 它使用多重处理,并且我们获得提交任务的进程池。 此池将任务分配给可用的进程并安排它们运行。
如何创建一个 ThreadPoolExecutor 或者 ProcessPoolExecutor?
在concurrent.futures
模块及其具体子类Executor
的帮助下,可以很容易地创建一个线程池或者进程池。 需要使用我们想要的池中的线程数构造一个ThreadPoolExecutor 或者 ProcessPoolExecutor
。 默认情况下,数字是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()
结果截图:
在上面的例子中,一个ThreadPoolExecutor
已经由5个线程构造而成。 然后,在提供消息之前等待2秒的任务被提交给线程池执行器。 从输出中可以看出,任务直到2
秒才完成,所以第一次调用done()
将返回False
。 2
秒后,任务完成,我们通过调用result()
方法得到future
的结果。
进程池 示例代码:
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()
实例化ThreadPoolExecutor 或者 ProcessPoolExecutor 之 上下文管理器
另一种实例化ThreadPoolExecutor
的方法是在上下文管理器的帮助下完成的。 它的工作方式与上例中使用的方法类似。 使用上下文管理器的主要优点是它在语法上看起来不错。 实例化可以在下面的代码的帮助下完成
with ThreadPoolExecutor(max_workers = 5) as executor
或者
with ProcessPoolExecutor(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/', 'https://www.yiibai.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.yiibai.com/' page is 168933 bytes
-
'http: //www.bbc.co.uk/' page is 283893 bytes
-
'http: //europe.wsj.com/' page is 938109 bytes
进程池:
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()
使用 map() 调度多任务,有序返回
使用map(),多个worker并发地从输入迭代器里取数据,处理,然后按顺序返回结果。
示例代码:
from concurrent import futures import time import random def task(n): time.sleep(random.randint(1, 10)) return n executor = futures.ThreadPoolExecutor(max_workers=3) results = executor.map(task, range(1, 10)) print('unprocessed results: {}'.format(results)) real_results = list(results) print('real results: {}'.format(real_results))
使用 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脚本的输出 :
进程池:
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()
多任务调度,无序返回
不断将任务submit到executor,返回future列表,使用as_completed无序产生每个任务的结果。
示例代码:
from concurrent import futures import time import random def task(n): time.sleep(random.randint(1, 10)) return n executor = futures.ThreadPoolExecutor(max_workers=3) future_list = [executor.submit(task, i) for i in range(1, 10)] for f in futures.as_completed(future_list): print(f.result())
何时使用ProcessPoolExecutor 和 ThreadPoolExecutor ?
现在我们已经学习了两个Executor
类 - ThreadPoolExecutor
和ProcessPoolExecutor
,我们需要知道何时使用哪个执行器。需要在受CPU限制的工作负载情况下选择ProcessPoolExecutor
,而在受I/O限制的工作负载情况下则需要选择ThreadPoolExecutor
。
如果使用ProcessPoolExecutor
,那么不需要担心GIL,因为它使用多处理。 而且,与ThreadPoolExecution
相比,执行时间会更少。
转载:https://blog.csdn.net/freeking101/article/details/97395745