concurrent.futures 官方文档：https://docs.python.org/3/library/concurrent.futures.html
concurrent.futures: 线程池, 并发的处理任务：https://www.h3399.cn/201906/703751.html

IO 密集型 vs 计算密集型：

IO密集型：读取文件，读取网络套接字频繁。
计算密集型：大量消耗CPU的数学与逻辑运算，也就是我们这里说的平行计算。

Python 因为其全局解释器锁 GIL 而无法通过线程实现真正的平行计算。但是 Python 的 concurrent.futures 模块可以利用 multiprocessing 实现真正的平行计算。

原理：concurrent.futures 会以子进程的形式，平行的运行多个 python 解释器，从而令 python 程序可以利用多核 CPU 来提升执行速度。由于子进程与主解释器相分离，所以他们的全局解释器锁也是相互独立的。每个子进程都能够完整的使用一个CPU 内核。

解释 2：concurrent.futures 中的 ProcessPoolExecutor类把工作分配给多个Python进程处理，因此，如果需要做CPU密集型处理，使用这个模块能绕开GIL，利用所有的CPU核心。
其原理是一个ProcessPoolExecutor创建了N个独立的Python解释器，N是系统上面可用的CPU核数。使用方法和ThreadPoolExecutor方法一样

1、Concurrent.futures 模块

标准库 concurrent.futures 是 python3.2+ 自带，python2 需要安装，它提供 ThreadPoolExecutor (线程池) 和 ProcessPoolExecutor (进程池) 两个类，实现了对 threading 和 multiprocessing 的更高级的抽象，对编写线程池/进程池提供支持。可以将相应的 tasks 直接放入线程池/进程池，不需要维护Queue来操心死锁的问题，线程池/进程池会自动调度。

打开 Concurrent.futures 模块路径，可以看到 python 文件如下：

_base.py 文件：

可以看到，有 Future类 和 Executor类。还有 as_completed 方法和 wait 方法

process.py 内容如下：

thread.py 和 process.py 内容差不多，这里只看 ThreadPoolExecutor

2、concurrent.futures 模块中的类

concurrent.futures

基础模块是 Executor 和 Future，
还有两个方法 as_completed 和 wait

2.1 concurrent.futures.as_completed 方法

concurrent.futures.as_completed()：返回已经执行完成的 Future 对象的列表。

as_completed() 方法是一个生成器，在没有任务完成的时候，会一直阻塞，除非设置了 timeout。

当有某个任务完成的时候，会 yield 这个任务，就能执行 for 循环下面的语句，然后继续阻塞住，循环到所有的任务结束。同时，先完成的任务会先返回给主线程。

# coding: utf-8
from concurrent.futures import ThreadPoolExecutor, as_completed
import timedef worker_func(page):time.sleep(page)print(f"worker ---> {page} finished")return pagedef main():with ThreadPoolExecutor(max_workers=5) as t:obj_list = []for page in range(1, 5):obj = t.submit(worker_func, page)obj_list.append(obj)for fd in as_completed(obj_list):data = fd.result()print(f"main: {data}")print("main 结束")if __name__ == '__main__':main()

示例：

import requests
from concurrent.futures import ThreadPoolExecutor as Pool
from concurrent.futures import as_completedURLS = ['https://www.baidu.com', 'https://qq.com', 'https://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(f"{str(f)} is running")for f in as_completed(future_task):try:ret = f.done()if ret:f_ret = f.result()print(f'{str(f)}, done, result: {f_ret.url}, {len(f_ret.content)}')except BaseException as be:f.cancel()print(be)"""
url不是按照顺序返回的，说明并发时，当访问某一个url时，如果没有得到返回结果，不会发生阻塞
"""

使用示例代码：

# -*- coding:utf-8 -*-import redis
from redis import WatchError
from concurrent.futures import ProcessPoolExecutorr = redis.Redis(host='127.0.0.1', port=6379)# 减库存函数, 循环直到减库存完成
# 库存充足, 减库存成功, 返回True
# 库存不足, 减库存失败, 返回Falsedef 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 Trueelse:return Falseexcept WatchError as ex:# 打印WatchError异常, 观察被watch锁住的情况print(ex)pipe.unwatch()def worker():while True:# 没有库存就退出if not reduce_stock():breakif __name__ == "__main__":# 设置库存为100r.set("stock:count", 100)# 多进程模拟多个客户端提交with ProcessPoolExecutor() as pool:for _ in range(10):pool.submit(worker)

2.2 concurrent.futures.wait 方法

wait(fs, timeout=None, return_when=ALL_COMPLETED)
wait 接受三个参数：

fs: 表示需要执行的序列
timeout: 等待的最大时间，如果超过这个时间即使线程未执行完成也将返回
return_when：wait 返回结果的条件，默认为 ALL_COMPLETED 全部执行完成再返回。如果采用默认的 ALL_COMPLETED，程序会阻塞直到线程池里面的所有任务都完成，再执行主线程。

返回值：

返回一个元组 " (已完成的 future 集合, 未完成的 future 集合) "

示例：

import time
import random
from concurrent.futures import ThreadPoolExecutor, waitdef func_test(int_1, int_2):sleep_second = random.randint(int_1, int_2)print(f'睡眠时间 {sleep_second}')time.sleep(sleep_second)passdef main():with ThreadPoolExecutor(max_workers=100) as tp_executor:future_task = [tp_executor.submit(func_test, 1, 5) for _ in range(100)]finished, doing = wait(future_task)print(f"finished ---> {len(finished)}")print(f"doing ---> {len(doing)}")if __name__ == '__main__':main()pass

示例：

from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED, ALL_COMPLETED
import timedef spider(page):time.sleep(page)print(f"crawl task{page} finished")return pagewith ThreadPoolExecutor(max_workers=5) as t:all_task = [t.submit(spider, page) for page in range(1, 5)]wait(all_task, return_when=FIRST_COMPLETED)print('finished')print(wait(all_task, timeout=2.5))

代码中返回的条件是：当完成第一个任务的时候，就停止等待，继续主线程任务
由于设置了延时，可以看到最后只有 task4 还在运行中

示例：

from concurrent.futures import Future
from concurrent.futures import ThreadPoolExecutor as Pool
from concurrent.futures import as_completed, wait
import requestsURLS = ['https://www.baidu.com', 'https://qq.com', 'https://sina.com']def task(url, timeout=3):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(f"正在运行 ---> {str(f)}")"""并且wait还有timeout和return_when两个参数return_when有三个常量 (默认是 ALL_COMPLETED)FIRST_COMPLETED 任何一个future_task执行完成时/取消时，该函数返回FIRST_EXCEPTION 任何一个future_task发生异常时，该函数返回，如果没有异常发生，等同于ALL_COMPLETED    ALL_COMPLETED 当所有的future_task执行完毕返回。"""finished, doing = wait(future_task, return_when="FIRST_COMPLETED")# finished, doing = wait(future_task, return_when="FIRST_EXCEPTION")# finished, doing = wait(future_task, return_when="ALL_COMPLETED")for item in finished:print(f"成功 ---> {item}")

2.3 Future 对象

future 可以理解为一个在未来完成的操作，这是异步编程的基础。通常情况下在遇到 IO 操作时会发生阻塞，cpu 不能做其他事情，而 future 的引入可以在等待的这段时间可以完成其他操作

future 是 concurrent.futures 模块和 asyncio 模块的重要组件。从 python3.4 开始标准库中有两个名为 Future 的类：

concurrent.futures.Future
asyncio.Future

这两个类的作用相同：两个 Future类的实例都表示可能完成或者尚未完成的延迟计算。与Twisted中的 Deferred 类、Tornado 框架中的 Future 类的功能类似。

这两种 future 都有.done()方法，这个方法不阻塞，返回值是布尔值，指明future链接的可调用对象是否已经执行。客户端代码通常不会询问future是否运行结束，而是会等待通知。因此两个Future类都有.add_done_callback()方法，这个方法只有一个参数，类型是可调用的对象，future运行结束后会调用指定的可调用对象。

.result()方法是在两个Future类中的作用相同：返回可调用对象的结果，或者重新抛出执行可调用的对象时抛出的异常。但是如果future没有运行结束，result方法在两个Futrue类中的行为差别非常大。对 concurrent.futures.Future实例来说，调用.result()方法会阻塞调用方所在的线程，直到有结果可返回，此时，result方法可以接收可选的timeout参数，如果在指定的时间内future没有运行完毕，会抛出TimeoutError异常。而 asyncio.Future.result方法不支持设定超时时间，在获取future结果最好使用yield from结构，但是concurrent.futures.Future不能这样做。

不管是 asyncio 还是 concurrent.futures.Future 都会有几个函数是返回 future，其他函数则是使用future，在最开始的例子中我们使用的Executor.map就是在使用future，返回值是一个迭代器，迭代器的__next__方法调用各个future的result方法，因此我们得到的是各个futrue的结果，而不是future本身，

注意：通常不应该自己创建 future，而是由并发框架 ( concurrent.futures 或 asyncio ) 实例化。
原因：future 表示终将发生的事情，而确定某件事情会发生的唯一方式是执行的时间已经安排好，因此只有把某件事情交给 concurrent.futures.Executor 子类处理时，才会创建concurrent.futures.Future 实例。如：Executor.submit() 方法的参数是一个可调用的对象，调用这个方法后会为传入的可调用对象排定时间，并返回一个 future，同时客户端代码不能应该改变 future 的状态，并发框架在 future 表示的延迟计算结束后会改变期物的状态，我们无法控制计算何时结束。

Future 对象源码

常用方法如下：

done()：如果当前线程已取消/已成功，返回True。这个方法不阻塞。
add_done_callback() ：这个方法只有一个参数，类型是可调用的对象，future 运行结束后会调用指定的可调用对象。
cancel()：如果当前线程正在执行并且不能取消则调用返回Flase。否则调用取消时返回 True
running()：如果当前的线程正在执行，则返回True
result()：返回所调用任务的返回值。方法会阻塞调用方所在的线程。
如果调用尚未完成，则此方法等待
如果等待超时，会抛出concurrent.futures.TimeoutError
如果没有指定超时时间，则等待无时间限制
如果在完成之前，取消了Future，则会引发CancelledError

示例：

# coding: utf-8
from concurrent.futures import ThreadPoolExecutor
import timedef worker_func(page):time.sleep(page)print(f"worker ---> {page} 完成")return pagewith ThreadPoolExecutor(max_workers=5) as t:  # 创建一个最大容纳数量为5的线程池task1 = t.submit(worker_func, 1)task2 = t.submit(worker_func, 2)  # 通过submit提交执行的函数到线程池中task3 = t.submit(worker_func, 3)print(f"task1: {task1.done()}")  # 通过done来判断线程是否完成print(f"task2: {task2.done()}")print(f"task3: {task3.done()}")print("睡眠 3s 后继续")time.sleep(3)print(f"task1: {task1.done()}")print(f"task2: {task2.done()}")print(f"task3: {task3.done()}")print(task1.result())  # 通过result来获取返回值

使用 with 语句，通过 ThreadPoolExecutor 构造实例，同时传入 max_workers 参数来设置线程池中最多能同时运行的线程数目。
使用 submit 函数来提交线程需要执行的任务到线程池中，并返回该任务的句柄。注意 submit() 不会阻塞，而是立即返回。
通过 done() 方法判断该任务是否结束。上面示例提交任务后立即判断任务状态，显示四个任务都未完成。在延时 3s 后，task1 和 task2 执行完毕，task3 仍在执行中。
使用 result() 方法可以获取任务的返回值。

2.4 Executer 对象

class concurrent.futures.Executor 是 Python concurrent.futures 模块的一个抽象类，它提供了异步执行调用的方法，但是它不能直接使用，只能通过它的两个子类 ThreadPoolExecutor 或者 ProcessPoolExecutor 进行调用。

ThreadPoolExecutor：线程池
ProcessPoolExecutor：进程池

何时使用 ThreadPoolExecutor 和 ProcessPoolExecutor ？如果是在受CPU限制的工作负载情况下则选择 ProcessPoolExecutor，如果是而在受 I/O 限制的工作负载情况下则需要选择ThreadPoolExecutor。使用 ProcessPoolExecutor，那么不需要担心 GIL，因为它使用多处理。而且，与ThreadPoolExecution相比，执行时间会更少。

源码截图：

函数说明：

submit(fn, *args, **kwargs)：提交一个可执行的回调 task，然后返回一个 Future 对象，它是一个未来可期的对象，通过它可以获取线程的状态。即
主线程(或进程)中可以获取某一个线程(进程)执行的状态
主线程可以获取某一个线程（或者任务的）的状态，以及返回值。
当一个线程完成的时候，主线程能够立即知道。
map(fn, *iterables, timeout=None, chunksize=1)
fn： 第一个参数 fn 是需要线程执行的函数；
iterables：第二个参数接受一个可迭代对象；
timeout： 设置每次异步操作的超时时间。但由于 map 是返回线程执行的结果，如果 timeout 小于线程执行时间会抛异常 TimeoutError。
返回值：返回一个 map()迭代器，在这个迭代器中，回调执行完成后所返回的结果是有序的。

**Executor.submit(fn, *args, kwargs)

当 submit() 任务时，会返回 Future对象。 Future对象有一个名为 done() 的方法，它告诉Future是否已经解决。有了这个，为这个特定的 Future 对象设定了一个值。当任务完成时，线程池执行器将该值设置为 Future 的对象。

示例代码：

# -*- coding:utf-8 -*-
from concurrent import futuresdef test(num):import timereturn time.ctime(), numwith futures.ThreadPoolExecutor(max_workers=1) as executor:future = executor.submit(test, 1)print(future.result())

示例代码：

from concurrent import futures
import time
import randomdef task(n):time.sleep(random.randint(1, 10))return nexecutor = futures.ThreadPoolExecutor(max_workers=3)
future = executor.submit(task, 5)
print('future: {}'.format(future))
result = future.result()
print('result: {}'.format(result))

**Executor.map(func, *iterables, timeout=None)**

使用 map 方法，无需提前使用 submit 方法，map 方法与 python 高阶函数 map 的含义相同，都是将序列中的每个元素都执行同一个函数。如果操作超时，会返回 raisesTimeoutError；如果不指定 timeout 参数，则不设置超时间。回调执行完成后所返回的结果是有序的。

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

示例：

# -*- coding:utf-8 -*-
from concurrent import futuresdef test(num):import timereturn time.ctime(), numdata = [1, 2, 3]
with futures.ThreadPoolExecutor(max_workers=1) as executor:for future in executor.map(test, data):print(future)

示例：

import time
import datetime
from concurrent.futures import (as_completed, ThreadPoolExecutor
)def print_msg(args):print(f'[{datetime.datetime.now().replace(microsecond=0)}] {args}')def worker_func(n):time.sleep(n)print_msg(f"睡眠{n}秒后返回")return ndef executor_map():""" 回调执行完成后所返回的结果是有序的 """task_list = [5, 4, 3, 2, 1]with ThreadPoolExecutor(max_workers=3) as executor:result_list = executor.map(worker_func, task_list)print_msg("executor_map")for result in result_list:print_msg(f"得到返回结果 ---> {result}")def executor_submit():task_list = [5, 4, 3, 2, 1]with ThreadPoolExecutor(max_workers=3) as executor:future_list = [executor.submit(worker_func, arg) for arg in task_list]for future in future_list:print_msg(f"future ---> {future}")done_list = as_completed(future_list)for done in done_list:print_msg(f"done ---> {done},  得到返回结果 ---> {done.result()}")if __name__ == '__main__':# executor_map()executor_submit()

Executor.shutdown(wait=True)

释放系统资源，在 Executor.submit() 或 Executor.map() 等异步操作后调用。使用 with 语句可以避免显式调用此方法。

示例：

import time
from concurrent.futures import (Future, ThreadPoolExecutor, ProcessPoolExecutor
)def return_future(msg):time.sleep(3)return msgpool = 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())pool.shutdown()
print("主线程")

2.5 ThreadPoolExecutor 对象

ThreadPoolExecutor类是 Executor子类，使用线程池执行异步调用。

class concurrent.futures.ThreadPoolExecutor(max_workers)，使用 max_workers 数目的线程池执行异步调用。

python3 标准库 concurrent.futures 比原 Thread 封装更高，利用 concurrent.futures.Future 来进行各种便捷的数据交互，包括处理异常，都在 result() 中再次抛出。

示例：

from concurrent.futures import ThreadPoolExecutor
from time import sleepdef task(message):sleep(2)return messagedef main():executor = ThreadPoolExecutor(5)future = executor.submit(task, "Completed")print(future.done())sleep(2)print(future.done())print(future.result())if __name__ == '__main__':main()

示例：

import concurrent.futures
import urllib.requestURLS = ['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)))

2.6. 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 futuresdef worker_func(num):import timereturn time.ctime(), numdef main(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(worker_func, 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(f'异常 : {future.exception()}')else:print(f'运行结果 : {future.result()}')if __name__ == '__main__':main(5, 1)pass

示例：

from concurrent.futures import ProcessPoolExecutor
from time import sleepdef task(message):sleep(2)return messagedef main():executor = ProcessPoolExecutor(5)future = executor.submit(task, ("Completed"))print(future.done())sleep(2)print(future.done())print(future.result())if __name__ == '__main__':main()

示例：

import requests
from concurrent.futures import ProcessPoolExecutor, as_completedURLS = ['https://www.bing.com/','https://www.google.com/','https://www.baidu.com/','https://www.tencent.com/',
]def load_url(url, timeout):try:resp = requests.get(url, timeout=timeout, verify=False)return resp.contentexcept BaseException as be:raise bedef main():with ProcessPoolExecutor(max_workers=5) as executor:future_to_url = {executor.submit(load_url, url, 5): url for url in URLS}for future in as_completed(future_to_url):url = future_to_url[future]try:data = future.result()except BaseException as be:print(f'异常 {url} {be}')else:print(f'{url} {len(data)}')if __name__ == '__main__':main()