原文来自开源中国

python标准库提供线程和多处理模块来编写相应的多线程/多进程代码，但当项目达到一定规模时，频繁地创建/销毁进程或线程是非常消耗资源的，此时我们必须编写自己的线程池/进程池来交换时间空间。但是从Python3.2开始，标准库为我们提供了并发的。Futures模块，它提供两个类：ThreadPool Executor和ProcessPool Executor。它实现线程和多处理的进一步抽象，并为编写线程池/进程池提供直接支持。

concurrent.futures模块的基础是Exectuor，Executor是一个抽象类，它不能被直接使用。但是它提供的两个子类ThreadPoolExecutor和ProcessPoolExecutor却是非常有用，顾名思义两者分别被用来创建线程池和进程池的代码。我们可以将相应的tasks直接放入线程池/进程池，不需要维护Queue来操心死锁的问题，线程池/进程池会自动帮我们调度。

我们先通过下面这段代码来了解一下线程池的概念

#example1.pyfromconcurrent.futuresimportThreadPoolExecutorimporttimedefreturn_future_result(message):time.sleep(2)returnmessagepool=ThreadPoolExecutor(max_workers=2)#创建一个最大可容纳2个task的线程池future1=pool.submit(return_future_result,("hello"))#往线程池里面加入一个taskfuture2=pool.submit(return_future_result,("world"))#往线程池里面加入一个taskprint(future1.done())#判断task1是否结束time.sleep(3)print(future2.done())#判断task2是否结束print(future1.result())#查看task1返回的结果print(future2.result())#查看task2返回的结果

让我们根据操作结果进行分析。我们使用submit方法将任务添加到线程池，submit返回一个将来的对象，这可以简单地理解为将来要完成的操作。在第一份印刷声明中，很明显我们的未来1由于时间的原因没有完成。睡眠（2），因为我们使用时间挂起了主线程。sleep（3），所以到第二个print语句时，线程池中的所有任务都已完成。

ziwenxie::~»pythonexample1.pyFalseTruehelloworld#在上述程序执行的过程中，通过ps命令我们可以看到三个线程同时在后台运行ziwenxie::~»ps-eLf|greppythonziwenxie8361755783613319:45pts/000:00:00pythonexample1.pyziwenxie8361755783620319:45pts/000:00:00pythonexample1.pyziwenxie8361755783630319:45pts/000:00:00pythonexample1.py

上面的代码我们也可以改写为进程池形式，api和线程池如出一辙，我就不罗嗦了。

#example2.pyfromconcurrent.futuresimportProcessPoolExecutorimporttimedefreturn_future_result(message):time.sleep(2)returnmessagepool=ProcessPoolExecutor(max_workers=2)future1=pool.submit(return_future_result,("hello"))future2=pool.submit(return_future_result,("world"))print(future1.done())time.sleep(3)print(future2.done())print(future1.result())print(future2.result())

下面是运行结果

ziwenxie::~»pythonexample2.pyFalseTruehelloworldziwenxie::~»ps-eLf|greppythonziwenxie8560755785603319:53pts/000:00:00pythonexample2.pyziwenxie8560755785630319:53pts/000:00:00pythonexample2.pyziwenxie8560755785640319:53pts/000:00:00pythonexample2.pyziwenxie8561856085610119:53pts/000:00:00pythonexample2.pyziwenxie8562856085620119:53pts/000:00:00pythonexample2.py使用map/wait来操作线程池/进程池

除了submit，Exectuor还为我们提供了map方法，和内建的map用法类似，下面我们通过两个例子来比较一下两者的区别。

使用submit操作回顾

#example3.pyimportconcurrent.futuresimporturllib.requestURLS=['http://httpbin.org','http://example.com/','https://api.github.com/']defload_url(url,timeout):withurllib.request.urlopen(url,timeout=timeout)asconn:returnconn.read()#Wecanuseawithstatementtoensurethreadsarecleaneduppromptlywithconcurrent.futures.ThreadPoolExecutor(max_workers=3)asexecutor:#StarttheloadoperationsandmarkeachfuturewithitsURLfuture_to_url={executor.submit(load_url,url,60):urlforurlinURLS}forfutureinconcurrent.futures.as_completed(future_to_url):url=future_to_url[future]try:data=future.result()exceptExceptionasexc:print('%rgeneratedanexception:%s'%(url,exc))else:print('%rpageis%dbytes'%(url,len(data)))

从运行结果可以看出，as_completed不是按照URLS列表元素的顺序返回的。

ziwenxie::~»pythonexample3.py'http://example.com/'pageis1270byte'https://api.github.com/'pageis2039bytes'http://httpbin.org'pageis12150bytes

使用map

#example4.pyimportconcurrent.futuresimporturllib.requestURLS=['http://httpbin.org','http://example.com/','https://api.github.com/']defload_url(url):withurllib.request.urlopen(url,timeout=60)asconn:returnconn.read()#Wecanuseawithstatementtoensurethreadsarecleaneduppromptlywithconcurrent.futures.ThreadPoolExecutor(max_workers=3)asexecutor:forurl,datainzip(URLS,executor.map(load_url,URLS)):print('%rpageis%dbytes'%(url,len(data)))

从运行结果可以看出，map是按照URLS列表元素的顺序返回的，并且写出的代码更加简洁直观，我们可以根据具体的需求任选一种。

ziwenxie::~»pythonexample4.py'http://httpbin.org'pageis12150bytes'http://example.com/'pageis1270bytes'https://api.github.com/'pageis2039bytes

第三种选择wait

wait方法接会返回一个tuple(元组)，tuple中包含两个set(集合)，一个是completed(已完成的)另外一个是uncompleted(未完成的)。使用wait方法的一个优势就是获得更大的自由度，它接收三个参数FIRST_COMPLETED, FIRST_EXCEPTION 和ALL_COMPLETE，默认设置为ALL_COMPLETED。

我们通过下面这个例子来看一下三个参数的区别

fromconcurrent.futuresimportThreadPoolExecutor,wait,as_completedfromtimeimportsleepfromrandomimportrandintdefreturn_after_random_secs(num):sleep(randint(1,5))return"Returnof{}".format(num)pool=ThreadPoolExecutor(5)futures=[]forxinrange(5):futures.append(pool.submit(return_after_random_secs,x))print(wait(futures))#print(wait(futures,timeout=None,return_when='FIRST_COMPLETED'))

如果采用默认的ALL_COMPLETED，程序会阻塞直到线程池里面的所有任务都完成。

ziwenxie::~»pythonexample5.pyDoneAndNotDoneFutures(done={<Futureat0x7f0b06c9bc88state=finishedreturnedstr>,<Futureat0x7f0b06cbaa90state=finishedreturnedstr>,<Futureat0x7f0b06373898state=finishedreturnedstr>,<Futureat0x7f0b06352ba8state=finishedreturnedstr>,<Futureat0x7f0b06373b00state=finishedreturnedstr>},not_done=set())

如果采用FIRST_COMPLETED参数，程序并不会等到线程池里面所有的任务都完成。

ziwenxie::~»pythonexample5.pyDoneAndNotDoneFutures(done={<Futureat0x7f84109edb00state=finishedreturnedstr>,<Futureat0x7f840e2e9320state=finishedreturnedstr>,<Futureat0x7f840f25ccc0state=finishedreturnedstr>},not_done={<Futureat0x7f840e2e9ba8state=running>,<Futureat0x7f840e2e9940state=running>})