创建进程池与线程池concurrent.futures模块的使用

标签：ssi   任务   阻塞   BMI   串行   ret   示例   fine   tor   

一、进程池。

当并发的任务数量远远大于计算机所能承受的范围，即无法一次性开启过多的任务数量就应该考虑去 限制进程数或线程数，从而保证服务器不会因超载而瘫痪。这时候就出现了进程池和线程池。

二、concurrent.futures模块介绍

concurrent.futures模块提供了高度封装的异步调用接口

ThreadPoolExecutor：线程池，提供异步调用

ProcessPoolExecutor：进程池，提供异步调用

Both implement the same interface, which is defined by the abstract Executor class

三、基本方法：

submit(fn, *args, **kwargs)：异步提交任务

map(func, *iterables, timeout=None, chunksize=1)：取代for循环submit的操作

shutdown(wait=True)：相当于进程池的pool.close()+pool.join()操作

• wait=True，等待池内所有任务执行完毕回收完资源后才继续
• wait=False，立即返回，并不会等待池内的任务执行完毕
• 但不管wait参数为何值，整个程序都会等到所有任务执行完毕
• submit和map必须在shutdown之前

result(timeout=None)：取得结果

add_done_callback(fn)：回调函数

done()：判断某一个线程是否完成

cancle()：取消某个任务

四、进程池代码实例——ProcessPoolExecutor

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import current_process
import time

def func(i):
    print(f'进程 {current_process().name} 正在执行任务 {i}')
    time.sleep(1)
    return i**2

if __name__ == '__main__':
    pool = ProcessPoolExecutor(4)  # 进程池只有4个进程
    lt = []
    for i in range(20):  # 假设执行20个任务
        future = pool.submit(func,i)   # func任务要做20次,4个进程负责完成这个20个任务
        # print(future.result())   # 如果没有结果就一直等待拿到结果，导致了所有任务都在串行
        lt.append(future)
    pool.shutdown() # 默认为True,关闭了池的入口，会等待所有的任务执行完，结束阻塞，
    for fu in lt:
        print(fu.result())  # 等待所有的任务都执行完了，一起把返回值打印出来

五、线程池代码示例——ThreadPoolExecutor

from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
import time

def func(i):
    print(f'线程 {currentThread().name} 正在执行任务 {i}')
    time.sleep(1)
    return i**2

if __name__ == '__main__':
    fool = ThreadPoolExecutor(4)  # 线程池里只有4个线程
    lt = []
    for i in range(20):
        futrue = fool.submit(func,i)   # func任务要做20次，4个线程负责完成这20次任务
        lt.append(futrue)
    fool.shutdown()  # 默认为True,关闭了池的入口，会等待所有的任务执行完，结束阻塞，
    for fu in lt:
        print(fu.result())   # 等待所有的任务都执行完了，一起把返回值打印出来

六、回调函数add_done_callback(fn)

提交任务的两种方式：
同步： 提交了一个任务，必须等任务执行完了（拿到返回值），才能执行下一行代码
异步： 提交了一个任务，不要等执行完了，可以直接执行下一行代码。

ps:进程和线程回调方法的使用写一块了，注释掉的是进程的使用。

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
from threading import currentThread
from multiprocessing import current_process
import time

def task(i):
    print(f'线程 {currentThread().name} 正在执行任务 {i}')
    # print(f'进程 {current_process().name} 正在执行任务 {i}')
    time.sleep(1)
    return i**2

def parse(futrue):
    # 处理拿到的结果
    print(futrue.result())

if __name__ == '__main__':
    pool = ThreadPoolExecutor(4)  # 线程池里只有4个线程
    # pool = ProcessPoolExecutor(4)  # 进程池里只有4个进程
    lt = []
    for i in range(20):
        futrue = pool.submit(task,i)  # task任务要做20次，分别由四个进程完成这20个任务
        futrue.add_done_callback(parse)
        # 为当前任务绑定一个函数，在当前任务执行结束的时候会触发这个函数
        # 会把futrue对象作为参数传给函数
        # 这个称之为回调函数，处理完了回来就调用这个函数。

跟上面线程池里的例子相比：回调函数的作用，不需要等待所有的任务执行完才打印返回值。每执行完一个任务直接打印结果，实现一个并发的效果，效率有所提升。

创建进程池与线程池concurrent.futures模块的使用

标签：ssi   任务   阻塞   BMI   串行   ret   示例   fine   tor   

原文地址：https://www.cnblogs.com/guapitomjoy/p/11564397.html