在并发中使用线程池

以前讲过多线程和多进程的优点与缺点，多线程相对于多进程更加节省资源，但是在cython中多线程处理cpu密集型任务表现不好，多进程虽然处理能力很强，但是浪费资源过多而且进程间通信也比较麻烦。因此，如何更好地使用多线程和多进程就是一个问题了，不过，有问题就会有办法，随后就有了池的概念，池可以理解为开放了固定数量的线程或者进程，关闭池之后就不会再接受新的线程或者进程，这样一来就可以防止过度地使用线程和进程，同时也节省了资源。

我们来学习一下线程池，进程池的使用方法和线程池是一样的。在python中有一个库concurrent，这个库里有一个模块futures，该模块提供了ThreadPoolExecutor和ProcessPoolExecutor类来支持线程池和进程池。这两个池是高级一点的池，为什么这么说呢？它们其实调用更低级的池，而更低级的池在multiprocessing库中。所以说，ThreadPoolExecutor和ProcessPoolExecutor使用起来更方便一些。

看一看如何使用线程池来并发：

""" 线程池使用方法
1.pool_executor = concurrent.futures.ThreadPoolExecutor(thread_count)
2.pool_executor.map()或者pool_executor.submit()
3.必要时获取结果Future类有回调函数
future = pool_executor.submit(fn)
future.add_done_callback(callback)the statistics of this file:
lines(count)    understand_level(h/m/l)    classes(count)    functions(count)    fields(count)
000000000099    ----------------------m    00000000000000    0000000000000004    ~~~~~~~~~~~~~
"""import time
from concurrent import futures__author__ = '与C同行'def wait_block_n_second(n, verbose):print('进入wait_block_n_second函数')time.sleep(n)print(f'{verbose}离开wait_block_n_second函数')return n+1def future_main():start_time = time.time()wait_executor = futures.ThreadPoolExecutor(3)futures_list = []for i in range(7):future = wait_executor.submit(wait_block_n_second, i+1, f'{i+1}s之后')futures_list.append(future)wait_executor.shutdown(wait=True)end_time = time.time()print(f'使用3个线程池来运行7个函数耗时:{end_time-start_time}s')for future in futures_list:print(future.result())def map_main():start_time = time.time()wait_executor = futures.ThreadPoolExecutor(3)results = wait_executor.map(wait_block_n_second,[1, 7, 3, 4, 5, 6, 2],['1s之后','7s之后','3s之后','4s之后','5s之后','6s之后','2s之后'])wait_executor.shutdown(wait=True)end_time = time.time()print(f'使用3个线程池来运行7个函数耗时:{end_time-start_time}s')for result in results:print(result)def map_future_main():futures_list = []def callback(future_func):futures_list.append(future_func)start_time = time.time()wait_executor = futures.ThreadPoolExecutor(14)for i in range(7):future = wait_executor.submit(wait_block_n_second, i + 1, f'{i + 1}s之后')future.add_done_callback(callback)results = wait_executor.map(wait_block_n_second,[1, 7, 3, 4, 5, 6, 2],['1s之后','7s之后','3s之后','4s之后','5s之后','6s之后','2s之后'])wait_executor.shutdown(wait=True)end_time = time.time()print(f'使用默认线程池来运行14个函数耗时:{end_time-start_time}s')for future in futures_list:print(future.result())for result in results:print(result)if __name__ == '__main__':print(f'当前时间:{time.ctime()}')print()future_main()map_main()map_future_main()

先来看一看submit这个方法的效果：

if __name__ == '__main__':print(f'当前时间:{time.ctime()}')print()future_main()

这里开了三个线程，总共耗时12s，三个线程干活并不是一个线程干活，所以不是(1+7)×7/2=28s。

看看map函数如何工作的：

if __name__ == '__main__':print(f'当前时间:{time.ctime()}')print()map_main()

结果如下：

map函数要讲一个地方，看它的函数签名map(self, func, *iterables, ...)，iterables参数带星号，所以函数变量多的时候要分成不同的迭代对象传给iterables变量。

submit和map也可以一起使用：

if __name__ == '__main__':print(f'当前时间:{time.ctime()}')print()map_future_main()

结果如下：


这里开了14个线程，所以只用了7s就运行完了所有的任务。submit函数会返回一个Future对象，Future对象可以用result方法得到结果，也可以用add_done_callback方法来实现任务完成时调用的函数。

喜欢python的朋友可以关注微信公众号“与C同行”学习更多知识：