07. 多线程之线程池

一、概述

线程池在系统启动时即创建大量空闲的线程，程序只要将一个函数提交给线程池，线程池就会启动一个空闲的线程来执行它。当该函数执行结束后，该线程并不会死亡，而是再次返回到线程池中变成空闲状态，等待执行下一个函数。
此外，使用线程池可以有效地控制系统中并发线程的数量。当系统中包含有大量的并发线程时，会导致系统性能急剧下降，甚至导致 Python 解释器崩溃，而线程池的最大线程数参数可以控制系统中并发线程的数量不超过此数。
官网：https://docs.python.org/dev/library/concurrent.futures.html
从Python3.2开始，标准库为我们提供了concurrent.futures模块，它提供了ThreadPoolExecutor和ProcessPoolExecutor两个类，实现了对threading和multiprocessing的进一步抽象（这里主要关注线程池），不仅可以帮我们自动调度线程，还可以做到：
1. 主线程可以获取某一个线程（或者任务的）的状态，以及返回值。
2. 当一个线程完成的时候，主线程能够立即知道。
3. 让多线程和多进程的编码接口一致。


concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor：线程池，提供异步调用

属性：max_workers， 线程池容量

1、submit(fn, *args, **kwargs)
异步提交任务

2、map(func, *iterables, timeout=None, chunksize=1) 
取代for循环submit的操作

3、shutdown(wait=True) 
相当于进程池的pool.close()+pool.join()操作
wait=True，等待池内所有任务执行完毕回收完资源后才继续
wait=False，立即返回，并不会等待池内的任务执行完毕
但不管wait参数为何值，整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前

4、result(timeout=None)
取得结果
5、add_done_callback(fn)
回调函数
6、done()
方法用于判定某个任务是否完成
7、cancel()
cancel方法用于取消某个任务,该任务没有放入线程池中才能取消成功

二、创建线程池示例

#!/usr/bin/env python
# coding:utf-8

from concurrent.futures import ThreadPoolExecutor
import time


# 参数times用来模拟网络请求的时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times


if __name__ == '__main__':
    executor = ThreadPoolExecutor(max_workers=3)
    # 通过submit函数提交执行的函数到线程池中，submit函数立即返回，不阻塞
    task1 = executor.submit(get_html, times=3)
    task2 = executor.submit(get_html, 2)
    # done方法用于判定某个任务是否完成
    print(task1.done())
    # cancel方法用于取消某个任务,该任务没有放入线程池中才能取消成功
    print(task2.cancel())
    time.sleep(4)
    print(task1.done())
    # result方法可以获取task的执行结果
    print(task1.result())

ThreadPoolExecutor构造实例的时候，传入max_workers参数来设置线程池中最多能同时运行的线程数目。
使用submit函数来提交线程需要执行的任务（函数名和参数）到线程池中，并返回该任务的句柄（类似于文件、画图），注意submit()不是阻塞的，而是立即返回。
通过submit函数返回的任务句柄，能够使用done()方法判断该任务是否结束。上面的例子可以看出，由于任务有2s的延时，在task1提交后立刻判断，task1还未完成，而在延时4s之后判断，task1就完成了。
使用cancel()方法可以取消提交的任务，如果任务已经在线程池中运行了，就取消不了。这个例子中，线程池的大小设置为3，任务已经在运行了，所以取消失败。如果改变线程池的大小为1，那么先提交的是task1，task2还在排队等候，这是时候就可以成功取消。
使用result()方法可以获取任务的返回值。查看内部代码，发现这个方法是阻塞的。

三、as_completed方法获线程执行状态

as_completed()方法是一个生成器，在没有任务完成的时候，会阻塞，在有某个任务完成的时候，会yield这个任务，就能执行for循环下面的语句，然后继续阻塞住，循环到所有的任务结束。从结果也可以看出，先完成的任务会先通知主线程。

#!/usr/bin/env python
# coding:utf-8

from concurrent.futures import ThreadPoolExecutor, as_completed
import time


def get_html(times):
    time.sleep(times)
    print(f'get page {times}s finished.')
    return times


if __name__ == '__main__':
    executor = ThreadPoolExecutor(max_workers=5)
    urls = [3, 1, 2, 4]
    all_task = (executor.submit(get_html, url) for url in urls)
    for future in as_completed(all_task):
        data = future.result()
        print(f'main: get page {data}s success.')

四、map方法

使用map方法，无需提前使用submit方法，map方法与python标准库中的map含义相同，都是将序列中的每个元素都执行同一个函数。上面的代码就是对urls的每个元素都执行get_html函数，并分配各线程池。可以看到执行结果与上面的as_completed方法的结果不同，输出顺序和urls列表的顺序相同，就算2s的任务先执行完成，也会先打印出3s的任务先完成，再打印2s的任务完成。

#!/usr/bin/env python
# coding:utf-8

from concurrent.futures import ThreadPoolExecutor
import time


def get_html(times):
    time.sleep(times)
    print(f'get page {times}s finished.')
    return times


if __name__ == '__main__':
    executor = ThreadPoolExecutor(max_workers=5)
    urls = [3, 2, 4, 1]
    for data in executor.map(get_html, urls):
        print(f'main: get page {data}s success.')

五、wait方法

wait方法可以让主线程阻塞，直到满足设定的要求。
wait方法接收3个参数，等待的任务序列、超时时间以及等待条件。等待条件return_when默认为ALL_COMPLETED，表明要等待所有的任务都结束。可以看到运行结果中，确实是所有任务都完成了，主线程才打印出main。等待条件还可以设置为FIRST_COMPLETED，表示第一个任务完成就停止等待。

#!/usr/bin/env python
# coding:utf-8

from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED, FIRST_COMPLETED
import time


def get_html(times):
    time.sleep(times)
    print(f'get page {times}s finished.')
    return times


if __name__ == '__main__':
    executor = ThreadPoolExecutor(max_workers=5)
    urls = [3, 4, 2, 1]
    all_task = (executor.submit(get_html, url) for url in urls)
    # wait(all_task, return_when=ALL_COMPLETED)
    wait(all_task, return_when=FIRST_COMPLETED)
    print('main')

参考：[python] ThreadPoolExecutor线程池