ThreadPoolExecutor线程池

从Python3.2开始,标准库为我们提供了concurrent.futures模块,它提供了ThreadPoolExecutor和ProcessPoolExecutor两个类,实现了对threading和multiprocessing的进一步

抽象(这里主要关注线程池),不仅可帮我们自动调度线程,还可以做到:

1、主线程可以获取某一个线程(或者任务的)的状态,以及返回值。

2、当一个线程完成的时候,主线程能够立即知道。

3、让多线程和多进程的编码接口一致。

获取任务是否完成,取消任务和获取任务的返回结果

from concurrent.futures import ThreadPoolExecutor
import time


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


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

执行结果:

done? False
cancel? False
线程2 get page 2 finished
线程1 get page 3 finished
done? True
result? 3

1、ThreadPoolExecutor构造实例的时候,传入max_workers参数来设置线程池中最多能同时运行的线程数目。

2、使用submit函数来提交线程需要执行的任务(函数名和参数)到线程池中,并返回该任务的句柄(类似于文件、画图),注意submit()不是阻塞的,而是立即返回。

3、通过submit函数返回的任务句柄,能够使用done()方法判断该任务是否结束。上面的例子可以看出,由于任务有2s的延时,在task1提交后立刻判断,task1还未完成,

而在延时4s之后判断,task1就完成了。

4、使用cancel()方法可以取消提交的任务,如果任务已经在线程池中运行了,就取消不了。这个例子中,线程池的大小设置为2,任务已经在运行了,所以取下失败。如果

改变线程池的大小为1,那么先提交的是task1,task2还在排队等候,这个时候可以成功取消。

5、使用result()方法可以获取任务的返回值。查看内部代码,发现这个方法是阻塞的。

as_completed方法一次取出所有任务的结果

from concurrent.futures import ThreadPoolExecutor, as_completed
import time


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


tp = ThreadPoolExecutor(max_workers=2)
lst = [3, 2, 4]
all_task = [tp.submit(get_html, (i)) for i in lst]

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

执行结果:

get page 2 finished
in main: get page 2 success
get page 3 finished
in main: get page 3 success
get page 4 finished
in main: get page 4 success

map

from concurrent.futures import ThreadPoolExecutor
import time


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


tp = ThreadPoolExecutor(max_workers=2)
lst = [3, 2, 4]
for i in tp.map(get_html, lst):
    print(f"in main: get page {i} success")

执行结果:

get page 2 finished
get page 3 finished
in main: get page 3 success
in main: get page 2 success
get page 4 finished
in main: get page 4 success

wait方法可以让主线程阻塞,知道满足设定的要求

from concurrent.futures import ThreadPoolExecutor
import time


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


tp = ThreadPoolExecutor(max_workers=2)
lst = [3, 2, 4]
for i in tp.map(get_html, lst):
    print(f"in main: get page {i} success")

执行结果:

get page 2 finished
get page 3 finished
in main: get page 3 success
in main: get page 2 success
get page 4 finished
in main: get page 4 success
原文地址:https://www.cnblogs.com/lilyxiaoyy/p/12409328.html