Python线程池模块ThreadPoolExecutor用法分析

对于需要执行大量I/O型任务，使用多线程可以有效提高程序性能的同时，也存在着线程创建与销毁所带来的额外开销、资源竞争和同步问题等问题。线程池技术可以有效地缓解这些问题。Python中线程池的实现有很多，其中“ThreadPoolExecutor”是Python3内置的线程池实现，本文将详细讲解其使用方法。

ThreadPoolExecutor常用方法

创建线程池

ThreadPoolExecutor的构造函数定义为：

ThreadPoolExecutor(max_workers=None, thread_name_prefix="", initializer=None, initargs=())

其中，max_workers限制线程池中该同时运行的线程数量。如果没有设置，线程池默认的线程数量为1个。如果max_workers为None或者为负数，将抛出ValueError异常。thread_name_prefix是一个可选的前缀字符串，用来区分不同线程池的线程名称。initializer是可选的一个回调函数，用来在每个工作线程中启动的一个可调用对象。initargs是initializer回调函数的可变参数。

例如创建一个具有5个线程的线程池：

from concurrent.futures import ThreadPoolExecutor
pool = ThreadPoolExecutor(max_workers=5)

提交任务到线程池

使用ThreadPoolExecutor.submit()方法将一个任务提交给线程池。该方法返回一个Future对象，其中可以通过result()等方法获取任务执行结果。

def func(arg1, arg2):
    # 执行任务
    pass

future = pool.submit(func, arg1, arg2)
result = future.result()

关闭线程池

当线程池不需要再执行新的任务时，应该关闭线程池，以释放所有线程并回收系统资源。可以通过ThreadPoolExecutor.shutdown()或ThreadPoolExecutor.shutdown(wait=True)方法来关闭线程池。

ThreadPoolExecutor.shutdown()会禁止提交新的任务，等待所有线程执行完当前任务后关闭线程池。
ThreadPoolExecutor.shutdown(wait=True)会禁止提交新的任务，并等待所有线程执行完当前任务后关闭线程池。参数wait接受一个布尔值，用来决定关闭线程池时是否等待所有线程执行完当前任务。

例如：

pool.shutdown(wait=True)

示例一：使用线程池处理I/O型任务

首先，我们定义一个模拟的I/O型任务：

import time

def io_task(task_id):
    print("task %d start" % (task_id))
    time.sleep(1)
    print("task %d end" % (task_id))
    return task_id

然后，我们创建线程池并提交任务：

pool = ThreadPoolExecutor(max_workers=5)
task_results = []
for i in range(10):
    future = pool.submit(io_task, i)
    task_results.append(future)

pool.shutdown(wait=True)

for task_result in task_results:
    print("task %d return %d" % (task_result.result(), task_result.done()))

上述程序开启了一个具有5个线程的线程池，然后循环提交了10个任务到线程池中。由于线程池中一次只能执行一定数量的线程，所以只会同时执行5个任务，其余任务将被放入等待队列中。线程池将逐个执行等待队列中的任务。

示例二：使用线程池处理CPU密集型任务

不过，对于一般的CPU密集型任务，使用多线程反而可能会降低程序效率，因此Python中多线程通常用于处理I/O型任务。为了验证这一点，我们定义一个模拟的CPU密集型任务：

def cpu_task(task_id):
    a = 0
    for i in range(1, 5000001):
        a += i
    print("task %d end" % (task_id))
    return task_id

然后，我们在主线程中依次调用该函数：

start_time = time.time()
for i in range(3):
    cpu_task(i)
end_time = time.time()
print("time-consuming：%.2f s" % (end_time - start_time))

CPU密集型任务在一个线程中依次执行，总共花费了约18秒的时间。接下来，我们将该任务交给一个具有5个线程的线程池：

pool = ThreadPoolExecutor(max_workers=5)
task_results = []
start_time = time.time()
for i in range(3):
    future = pool.submit(cpu_task, i)
    task_results.append(future)

pool.shutdown(wait=True)

for task_result in task_results:
    print("task %d return %d" % (task_result.result(), task_result.done()))
end_time = time.time()
print("time-consuming：%.2f s" % (end_time - start_time))

由于线程池中一次只能执行一定数量的线程，所以只会同时执行5个任务，其余任务将被放入等待队列中。线程池将逐个执行等待队列中的任务。实验结果表明，该程序花费的时间与在一个线程中依次执行的时间基本相同。如果计算时间非常短，线程池的开销可能会反而增大。因此，在执行CPU密集型任务时，尽量不要使用多线程。