concurrent.futures模块简单介绍(线程池,进程池)
一、基类executor
executor类是threadpoolexecutor 和processpoolexecutor 的基类。它为我们提供了如下方法:
submit(fn, *args, **kwargs):提交任务。以 fn(*args **kwargs) 方式执行并返回 future 对像。
fn:函数地址。
*args:位置参数。
**kwargs:关键字参数。
map(func, *iterables, timeout=none, chunksize=1):
func:函数地址。
iterables:一个可迭代对象,以迭代的方式将参数传递给函数。
timeout:这个参数没弄明白,如果是none等待所有进程结束。
chunksize:使用 processpoolexecutor 时,这个方法会将 iterables 分割任务块,并作为独立的任务提交到执行池中。这些块的数量可以由 chunksize 指定设置。 对很长的迭代器来说,设置chunksize 值比默认值 1 能显著地提高性能。 chunksize 对 threadpoolexecutor 没有效果。
shutdown(wait=true):如果为true会等待线程池或进程池执行完成后释放正在使用的资源。如果 wait 为 false,将立即返回,所有待执行的期程完成执行后会释放已分配的资源。 不管 wait 的值是什么,整个 python 程序将等到所有待执行的期程完成执行后才退出。
二、线程池对象
threadpoolexecutor 是 executor 的子类,下面介绍threadpoolexecutor 的参数。
class concurrent.futures.threadpoolexecutor(max_workers=none, thread_name_prefix='', initializer=none, initargs=()):
max_workers:线程池的数量。
thread_name_prefix:线程名前缀。默认线程名threadpoolexecutor-线程数
initializer:一个函数或方法,在启用线程前会调用这个函数(给线程池添加额外任务)。
initargs :以元祖的方式给initializer中的函数传递参数。
这里需要说明的是除了max_workers这个参数外其它三个参数基本很少用。max_workers很好理解就是线程池的数量。
下面来说initializer和initargs 这两个奇怪的家伙。
示例一:
from concurrent.futures import threadpoolexecutor def work(): print('工作线程') def test(num): print('test:',num) executor = threadpoolexecutor(max_workers=2,initializer=test(7)) # 开启2个线程 initializer指定参数test(7) executor.submit(work) executor.submit(work) # 打印内容如下 test: 7 工作线程 工作线程
示例二:
from concurrent.futures import threadpoolexecutor def work(): print('工作线程') def test(num): print('test:',num) executor = threadpoolexecutor(max_workers=2,initializer=test,initargs=(7,)) # 这里我们使用initargs=(7,)的方式给test传递参数。 executor.submit(work) executor.submit(work) # 打印内容如下 test: 7 工作线程 工作线程 test: 7
通过示例一和示例二我们可以发现initializer=test(7)时,test函数只被调用了1次,当initializer=test,initargs=(7,)时,test被调用了2次。具体原因没有去分析。感觉没什么用。以后有时间看看源码在补上。
三、进程池对象
processpoolexecutor 也是 executor 的子类,下面是processpoolexecutor 参数介绍:
class concurrent.futures.processpoolexecutor(max_workers=none, mp_context=none, initializer=none, initargs=())
max_workers:工作进程数。如果 max_workers 为 none 或未给出,它将默认为机器的处理器个数。 如果 max_workers 小于等于 0,则将引发 valueerror。 在 windows 上,max_workers 必须小于等于 61,否则将引发 valueerror。 如果 max_workers 为 none,则所选择的默认最多为 61,即使存在更多处理器。
mp_context :可以是一个多进程上下文或是 none。 它将被用来启动工作进程。 如果 mp_context 为 none 或未给出,将使用默认的多进程上下文。
initializer:一个函数或方法,在启用线程前会调用这个函数。
initargs :以元祖的方式给initializer中的函数传递参数。
关于说initializer和initargs 与threadpoolexecutor 类似这里不多说了。
四、创建线程池
from concurrent.futures import threadpoolexecutor import time def work(num): time.sleep(1) print('工作线程:',num) if __name__ == '__main__': executor = threadpoolexecutor(max_workers=5) # 创建线程池,数量为5 for i in range(5): executor.submit(work, i) print('主线程') # 打印内容如下 主线程 工作线程: 0 工作线程: 1 工作线程: 2 工作线程: 3 工作线程: 4
# 使用shutdown等待所有线程结束后在打印主线程 from concurrent.futures import threadpoolexecutor import time def work(num): time.sleep(1) print('工作线程:',num) if __name__ == '__main__': executor = threadpoolexecutor(max_workers=5) # 创建线程池,数量为5 for i in range(5): executor.submit(work, i) executor.shutdown(wait=true) # 等待线程池结束 print('主线程') # 打印内容如下 工作线程: 0 工作线程: 1 工作线程: 2 工作线程: 3 工作线程: 4 主线程
如果想要在线程执行的过程中添加额外的功能,可以使用initializer参数,如下:
from concurrent.futures import threadpoolexecutor def work(num): print('工作线程:',num) def test(num): print('额外任务:',num) if __name__ == '__main__': executor = threadpoolexecutor(max_workers=5,initializer=test,initargs=(7,)) # 添加额外任务 for i in range(5): executor.submit(work, i) executor.shutdown(wait=true) print('主线程') # 打印内容如下 额外任务: 7 工作线程: 0 额外任务: 7 工作线程: 1 额外任务: 7 工作线程: 2 额外任务: 7 工作线程: 3 额外任务: 7 工作线程: 4 主线程
五、进程池
进程池与线程池用法基本一致,只是名字和实现不一样而已。
from concurrent.futures import processpoolexecutor import time def work(num): time.sleep(1) print('工作进程:',num) if __name__ == '__main__': executor = processpoolexecutor(max_workers=5) # 创建进程池,数量为5 for i in range(5): executor.submit(work, i) print('主线程') # 打印内容如下 主线程 工作进程: 0 工作进程: 1 工作进程: 2 工作进程: 3 工作进程: 4 # 使用shutdown等待所有线程结束后在打印主线程 from concurrent.futures import processpoolexecutor import time def work(num): time.sleep(1) print('工作进程:',num) if __name__ == '__main__': executor = processpoolexecutor(max_workers=5) # 创建进程池,数量为5 for i in range(5): executor.submit(work, i) executor.shutdown(wait=true) # 等待进程池结束 print('主线程') # 打印内容如下 工作进程: 0 工作进程: 1 工作进程: 2 工作进程: 3 工作进程: 4 主线程
如果想要在线程执行的过程中添加额外的功能,可以使用initializer参数,如下:
from concurrent.futures import processpoolexecutor def work(num): print('工作进程:',num) def test(num): print('额外任务:',num) if __name__ == '__main__': executor = processpoolexecutor(max_workers=5,initializer=test,initargs=(7,)) # 添加额外任务 for i in range(5): executor.submit(work, i) executor.shutdown(wait=true) print('主线程') # 打印内容如下 额外任务: 7 工作进程: 0 工作进程: 1 工作进程: 2 工作进程: 3 工作进程: 4 额外任务: 7 额外任务: 7 额外任务: 7 额外任务: 7 主线程
未完待续