0x02 线程池

线程池

线程对象的创建会占用资源, 管理也很麻烦. 对于需要创建很多线程的场景, 使用线程池来简化线程的管理, 并且对于执行完任务的线程输入新的数据, 重新利用, 从而减少了线程对象创建和回收的资源和运算成本, 同时也能够控制线程的规模, 防止无节制的创建新的线程, 造成系统的问题.

一般来说, 线程池只适用于处理I/O密集型的任务.

使用concurrent.futures中的ThreadPoolExecutor创建一个线程池. 创建方法如下, 需要制定线程池最大容纳的线程数量.

pool = ThreadPoolExecutor(2)

有两种启动令线程池启动线程方法.

pool.map

指定一个函数, 所有的线程都运行同一个函数, 每个线程的输入组成一个列表, 作为map函数的另一个输入.

def return_future_result(msg):
    time.sleep(2)
    print(threading.current_thread())
    return msg

result = pool.map(return_future_result, ["test1", "test2", "test3"])
result

输出为:

<generator object Executor.map.<locals>.result_iterator at 0x00000250E0EB1E08>

<Thread(ThreadPoolExecutor-0_0, started daemon 12088)>
<Thread(ThreadPoolExecutor-0_1, started daemon 13252)>
<Thread(ThreadPoolExecutor-0_0, started daemon 12088)>

所有输入运行的结果将会存放在result中, 可以看到map函数返回的是一个生成器对象. 查看生成器里面的内容:

[t for t in result]

结果为:

['test1', 'test2', 'test3']

pool.submit

也可以使用线程池的submit方法提交一个任务, 线程池将会启动一个线程来处理这个任务, 如果有空余线程的话, 是立即启动的.

r1 = pool.submit(return_future_result, args=("hello",))
r2 = pool.submit(return_future_result, ("world",))
r1, r2

输出为:

(<Future at 0x250e0ee1198 state=pending>,
 <Future at 0x250e0ee1240 state=pending>)

 <Thread(ThreadPoolExecutor-0_0, started daemon 12088)>
 <Thread(ThreadPoolExecutor-0_1, started daemon 13252)>

可以看到结果是以Future对象的形式返回的. Future对象的作用很多, 后面专门分析. 在此, 我们可以通过Future的done()方法查看任务是否执行完毕:

r1.done(), r2.done()

结果为:

(True, True)

可以通过result()方法查看函数返回的结果:

r1.result(), r2.result()

结果为:

('hello', 'world')

注意, .result()操作会造成阻塞, 如果任务在线程中没有执行完毕, 在主线程中就调用了result方法, 主线程就会被阻塞住, 直到对应函数已经由线程池执行完毕并返回了结果为止.

参考资料

Python 并发编程之线程池/进程池