[python] ThreadPoolExecutor线程池 python 线程池

初识

Python中已经有了threading模块，为何还须要线程池呢，线程池又是什么东西呢？在介绍线程同步的信号量机制的时候，举得例子是爬虫的例子，须要控制同时爬取的线程数，例子中建立了20个线程，而同时只容许3个线程在运行，可是20个线程都须要建立和销毁，线程的建立是须要消耗系统资源的，有没有更好的方案呢？其实只须要三个线程就好了，每一个线程各分配一个任务，剩下的任务排队等待，当某个线程完成了任务的时候，排队任务就能够安排给这个线程继续执行。

这就是线程池的思想（固然没这么简单），可是本身编写线程池很难写的比较完美，还须要考虑复杂状况下的线程同步，很容易发生死锁。从Python3.2开始，标准库为咱们提供了concurrent.futures模块，它提供了ThreadPoolExecutor和ProcessPoolExecutor两个类，实现了对threading和multiprocessing的进一步抽象（这里主要关注线程池），不只能够帮咱们自动调度线程，还能够作到：

1. 主线程能够获取某一个线程（或者任务的）的状态，以及返回值。
2. 当一个线程完成的时候，主线程可以当即知道。
3. 让多线程和多进程的编码接口一致。

实例

简单使用

from concurrent.futures import ThreadPoolExecutor
import time

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

executor = ThreadPoolExecutor(max_workers=2)
# 经过submit函数提交执行的函数到线程池中，submit函数当即返回，不阻塞
task1 = executor.submit(get_html, (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())

# 执行结果
# False  # 代表task1未执行完成
# False  # 代表task2取消失败，由于已经放入了线程池中
# get page 2s finished
# get page 3s finished
# True  # 因为在get page 3s finished以后才打印，因此此时task1必然完成了
# 3     # 获得task1的任务返回值

 

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

as_completed

上面虽然提供了判断任务是否结束的方法，可是不能在主线程中一直判断啊。有时候咱们是得知某个任务结束了，就去获取结果，而不是一直判断每一个任务有没有结束。这是就可使用as_completed方法一次取出全部任务的结果。

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

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

executor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并非真的url
all_task = [executor.submit(get_html, (url)) for url in urls]

for future in as_completed(all_task):
    data = future.result()
    print("in main: get page {}s success".format(data))

# 执行结果
# get page 2s finished
# in main: get page 2s success
# get page 3s finished
# in main: get page 3s success
# get page 4s finished
# in main: get page 4s success

 

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

map

除了上面的as_completed方法，还可使用executor.map方法，可是有一点不一样。

from concurrent.futures import ThreadPoolExecutor
import time

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

executor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并非真的url

for data in executor.map(get_html, urls):
    print("in main: get page {}s success".format(data))
# 执行结果
# get page 2s finished
# get page 3s finished
# in main: get page 3s success
# in main: get page 2s success
# get page 4s finished
# in main: get page 4s success

 

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

wait

wait方法可让主线程阻塞，直到知足设定的要求。

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

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

executor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并非真的url
all_task = [executor.submit(get_html, (url)) for url in urls]
wait(all_task, return_when=ALL_COMPLETED)
print("main")
# 执行结果 
# get page 2s finished
# get page 3s finished
# get page 4s finished
# main

 

wait方法接收3个参数，等待的任务序列、超时时间以及等待条件。等待条件 return_when默认为 ALL_COMPLETED，代表要等待全部的任务都结束。能够看到运行结果中，确实是全部任务都完成了，主线程才打印出 main。等待条件还能够设置为 FIRST_COMPLETED，表示第一个任务完成就中止等待。