1. concurrent.futures库简介
　　python标准库未我们提供了threading和mutiprocessing模块实现异步多线程/多进程功能。从python3.2版本开始，标准库又为我们提供了concurrent.futures模块来实现线程池和进程池功能，实现了对threading和mutiprocessing模块的高级抽象，更大程度上方便了我们python程序员。
　　concurrent.futures模块提供了ThreadPoolExecutor和ProcessPoolExecutor两个类
（1）看下来个类的继承关系和关键属性
复制代码
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
print(‘ThreadPoolExecutor继承关系：’,ThreadPoolExecutor.__mro__)
print(‘ThreadPoolExecutor属性：’,[attr for attr in dir(ThreadPoolExecutor) if not attr.startswith(‘_’)])
print(‘ProcessPoolExecutor继承关系：’,ProcessPoolExecutor.__mro__)
print(‘ThreadPoolExecutor属性：’,[attr for attr in dir(ProcessPoolExecutor) if not attr.startswith(‘_’)])
复制代码
　　都继承自futures._base.Executor类，拥有三个重要方法map、submit和shutdow，这样看起来就很简单了
（2）再看下futures._base.Executor基类实现
View Code
　　提供了map、submit、shutdow和with方法，下面首先对这个几个方法的使用进行说明
回到顶部
2. map函数
　　函数原型：def map(self, fn, *iterables, timeout=None, chunksize=1)
　　map函数和python自带的map函数用能类型，只不过该map函数从迭代器获取参数后异步执行，timeout用于设置超时时间
　　参数chunksize的理解：
The size of the chunks the iterable will be broken into
before being passed to a child process. This argument is only
used by ProcessPoolExecutor; it is ignored by ThreadPoolExecutor.
　　例：
复制代码
from concurrent.futures import ThreadPoolExecutor
import time
import requests
def download(url):
headers = {‘User-Agent’:’Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:63.0) Gecko/20100101 Firefox/63.0′,
‘Connection’:’keep-alive’,
‘Host’:’example.webscraping.com’}
response = requests.get(url, headers=headers)
return(response.status_code)
if __name__ == ‘__main__’:
urllist = [‘http://example.webscraping.com/places/default/view/Afghanistan-1’,
‘http://example.webscraping.com/places/default/view/Aland-Islands-2’]
pool = ProcessPoolExecutor(max_workers = 2)
start = time.time()
result = list(pool.map(download, urllist))
end = time.time()
print(‘status_code:’,result)
print(‘使用多线程–timestamp:{:.3f}’.format(end-start))
复制代码
回到顶部
3. submit函数
　　函数原型：def submit(self, fn, *args, **kwargs)
　　fn：需要异步执行的函数
　　args、kwargs：函数传递的参数
　　例：下例中future类的使用和as_complete后面介绍
复制代码
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor,as_completed
import time
import requests
def download(url):
headers = {‘User-Agent’:’Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:63.0) Gecko/20100101 Firefox/63.0′,
‘Connection’:’keep-alive’,
‘Host’:’example.webscraping.com’}
response = requests.get(url, headers=headers)
return response.status_code
if __name__ == ‘__main__’:
urllist = [‘http://example.webscraping.com/places/default/view/Afghanistan-1’,
‘http://example.webscraping.com/places/default/view/Aland-Islands-2’]
start = time.time()
pool = ProcessPoolExecutor(max_workers = 2)
futures = [pool.submit(download,url) for url in urllist]
for future in futures:
print(‘执行中:%s, 已完成:%s’ % (future.running(), future.done()))
print(‘#### 分界线 ####’)
for future in as_completed(futures, timeout=2):
print(‘执行中:%s, 已完成:%s’ % (future.running(), future.done()))
print(future.result())
end = time.time()
print(‘使用多线程–timestamp:{:.3f}’.format(end-start))
复制代码
　　输出：
　　
回到顶部
4. shutdown函数
　　函数原型：def shutdown(self, wait=True)
　　此函数用于释放异步执行操作后的系统资源
　　由于_base.Executor类提供了上下文方法，将shutdown封装在了__exit__中，若使用with方法，将不需要自己进行资源释放
with ProcessPoolExecutor(max_workers = 2) as pool:
回到顶部
5. Future类
　　submit函数返回Future对象，Future类提供了跟踪任务执行状态的方法：
　　future.running()：判断任务是否执行
　　futurn.done：判断任务是否执行完成
　　futurn.result()：返回函数执行结果
复制代码
futures = [pool.submit(download,url) for url in urllist]
for future in futures:
print(‘执行中:%s, 已完成:%s’ % (future.running(), future.done()))
print(‘#### 分界线 ####’)
for future in as_completed(futures, timeout=2):
print(‘执行中:%s, 已完成:%s’ % (future.running(), future.done()))
print(future.result())
复制代码
　　as_completed方法传入futures迭代器和timeout两个参数
　　默认timeout=None，阻塞等待任务执行完成，并返回执行完成的future对象迭代器，迭代器是通过yield实现的。
　　timeout>0，等待timeout时间，如果timeout时间到仍有任务未能完成，不再执行并抛出异常TimeoutError
回到顶部
6. 回调函数
　　Future类提供了add_done_callback函数可以自定义回调函数：
复制代码
def add_done_callback(self, fn):
“””Attaches a callable that will be called when the future finishes.
Args:
fn: A callable that will be called with this future as its only
argument when the future completes or is cancelled. The callable
will always be called by a thread in the same process in which
it was added. If the future has already completed or been
cancelled then the callable will be called immediately. These
callables are called in the order that they were added.
“””
with self._condition:
if self._state not in [CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED]:
self._done_callbacks.append(fn)
return
fn(self)
复制代码
　　例子：
复制代码
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor,as_completed
import time
import requests
def download(url):
headers = {‘User-Agent’:’Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:63.0) Gecko/20100101 Firefox/63.0′,
‘Connection’:’keep-alive’,
‘Host’:’example.webscraping.com’}
response = requests.get(url, headers=headers)
return response.status_code
def callback(future):
print(future.result())
if __name__ == ‘__main__’:
urllist = [‘http://example.webscraping.com/places/default/view/Afghanistan-1’,
‘http://example.webscraping.com/places/default/view/Aland-Islands-2’,
‘http://example.webscraping.com/places/default/view/Albania-3’,
‘http://example.webscraping.com/places/default/view/Algeria-4’,
‘http://example.webscraping.com/places/default/view/American-Samoa-5’]
start = time.time()
with ProcessPoolExecutor(max_workers = 2) as pool:
futures = [pool.submit(download,url) for url in urllist]
for future in futures:
print(‘执行中:%s, 已完成:%s’ % (future.running(), future.done()))
print(‘#### 分界线 ####’)
for future in as_completed(futures, timeout=5):
future.add_done_callback(callback)
print(‘执行中:%s, 已完成:%s’ % (future.running(), future.done()))
end = time.time()
print(‘使用多线程–timestamp:{:.3f}’.format(end-start))
复制代码
回到顶部
7. wait函数
　　函数原型：def wait(fs, timeout=None, return_when=ALL_COMPLETED)
View Code
　　wait方法返回一个中包含两个元组，元组中包含两个集合（set）,一个是已经完成的（completed）,一个是未完成的（uncompleted）
　　它接受三个参数，重点看下第三个参数：
　　FIRST_COMPLETED：Return when any future finishes or iscancelled.　
　　FIRST_EXCEPTION：Return when any future finishes by raising an exception，
　　　　　　　　　　　　 If no future raises an exception then it is equivalent to ALL_COMPLETED.
　　ALL_COMPLETED：Return when all futures finish or are cancelled
　　例：
复制代码
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor,\
as_completed,wait,ALL_COMPLETED, FIRST_COMPLETED, FIRST_EXCEPTION
import time
import requests
def download(url):
headers = {‘User-Agent’:’Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:63.0) Gecko/20100101 Firefox/63.0′,
‘Connection’:’keep-alive’,
‘Host’:’example.webscraping.com’}
response = requests.get(url, headers=headers)
return response.status_code
if __name__ == ‘__main__’:
urllist = [‘http://example.webscraping.com/places/default/view/Afghanistan-1’,
‘http://example.webscraping.com/places/default/view/Aland-Islands-2’,
‘http://example.webscraping.com/places/default/view/Albania-3’,
‘http://example.webscraping.com/places/default/view/Algeria-4’,
‘http://example.webscraping.com/places/default/view/American-Samoa-5’]
start = time.time()
with ProcessPoolExecutor(max_workers = 2) as pool:
futures = [pool.submit(download,url) for url in urllist]
for future in futures:
print(‘执行中:%s, 已完成:%s’ % (future.running(), future.done()))
print(‘#### 分界线 ####’)
completed, uncompleted = wait(futures, timeout=2, return_when=FIRST_COMPLETED)
for cp in completed:
print(‘执行中:%s, 已完成:%s’ % (cp.running(), cp.done()))
print(cp.result())
end = time.time()
print(‘使用多线程–timestamp:{:.3f}’.format(end-start))
来自：https://www.cnblogs.com/xiaobingqianrui/p/10058137.html