python中ThreadPoolExecutor线程池和ProcessPoolExecutor进程池怎么使用

这篇文章主要介绍了python中ThreadPoolExecutor线程池和ProcessPoolExecutor进程池怎么使用的相关知识，内容详细易懂，操作简单快捷，具有一定借鉴价值，相信大家阅读完这篇Python中ThreadPoolExecutor线程池和ProcessPoolExecutor进程池怎么使用文章都会有所收获，下面我们一起来看看吧。

1、ThreadPoolExecutor多线程

<1>为什么需要线程池呢?

• 对于io密集型，提高执行的效率。

• 线程的创建是需要消耗系统资源的。

所以线程池的思想就是：每个线程各自分配一个任务，剩下的任务排队等待，当某个线程完成了任务的时候，排队任务就可以安排给这个线程继续执行。

<2>标准库concurrent.futures模块

它提供了ThreadPoolExecutor和ProcessPoolExecutor两个类，
分别实现了对threading模块和multiprocessing模块的进一步抽象。

不仅可以帮我们自动调度线程，还可以做到:

• 主线程可以获取某一个线程(或者任务)的状态，以及返回值

• 当一个线程完成的时候，主线程能够立即知道

• 让多线程和多进程的编码接口一致

<3>简单使用

# -*-coding:utf-8 -*-from concurrent.futures import ThreadPoolExecutorimport time# 参数times用来模拟网络请求时间def get_html(times):    print("get page {}s finished".fORMat(times))   return times# 创建线程池# 设置线程池中最多能同时运行的线程数目，其他等待executor = ThreadPoolExecutor(max_workers=2)# 通过submit函数提交执行的函数到线程池中，submit函数立即返回,不阻塞# task1和task2是任务句柄task1 = executor.submit( get_html, (2) )task2 = executor.submit( get_html, (3) )# done()方法用于判断某个任务是否完成，bool型，完成返回True，没有完成返回Falseprint( task1.done() )# cancel()方法用于取消某个任务，该任务没有放到线程池中才能被取消，如果已经放进线程池子中，则不能被取消# bool型，成功取消了返回True，没有取消返回Falseprint( task2.cancel() )# result()方法可以获取task的执行结果，前提是get_html()函数有返回值print( task1.result() )print( task2.result() )# 结果：# get page 3s finished# get page 2s finished# True# False# 2# 3

ThreadPoolExecutor类在构造实例的时候，传入max_workers参数来设置线程池中最多能同时运行的线程数目
使用submit()函数来提交线程需要执行任务(函数名和参数)到线程池中，并返回该任务的句柄，

注意：submit()不是阻塞的，而是立即返回。

通过submit()函数返回的任务句柄，能够使用done()方法判断该任务是否结束，使用cancel()方法来取消，使用result()方法可以获取任务的返回值，查看内部代码，发现该方法是阻塞的

<4>as_completed（一次性获取所有的结果）

上面虽然提供了判断任务是否结束的方法，但是不能在主线程中一直判断，有时候我们是得知某个任务结束了，就去获取结果，而不是一直判断每个任务有没有结束。这时候就可以使用as_completed方法一次取出所有任务的结果。

# -*-coding:utf-8 -*-from concurrent.futures import ThreadPoolExecutor, as_completedimport time# 参数times用来模拟网络请求时间def get_html(times):    time.sleep(times)    print("get page {}s finished".format(times))    return times# 创建线程池子# 设置最多2个线程运行，其他等待executor = ThreadPoolExecutor(max_workers=2)urls = [3,2,4]# 一次性把所有的任务都放进线程池，得到一个句柄，但是最多只能同时执行2个任务all_task = [ executor.submit(get_html,(each_url)) for each_url in urls ] for future in as_completed( all_task ):    data = future.result()    print("in main:get page {}s success".format(data))# 结果# get page 2s finished# in main:get page 2s success# get page 3s finished# in main:get page 3s success# get page 4s finished# in main:get page 4s success# 从结果可以看到，并不是先传入哪个url，就先执行哪个url，没有先后顺序

<5>map()方法

除了上面的as_completed()方法，还可以使用execumap方法。但是有一点不同，使用map方法，不需提前使用submit方法，
map方法与python标准库中的map含义相同，都是将序列中的每个元素都执行同一个函数。上面的代码就是对urls列表中的每个元素都执行get_html()函数，并分配各线程池。可以看到执行结果与上面的as_completed方法的结果不同，输出顺序和urls列表的顺序相同，就算2s的任务先执行完成，也会先打印出3s的任务先完成，再打印2s的任务完成

# -*-coding:utf-8 -*-from concurrent.futures import ThreadPoolExecutor,as_completedimport time# 参数times用来模拟网络请求时间def get_html(times):    time.sleep(times)    print("get page {}s finished".format(times))    return times# 创建线程池子# 设置最多2个线程运行，其他等待executor = ThreadPoolExecutor(max_workers=2)urls = [3,2,4]for result in executor.map(get_html, urls):    print("in main:get page {}s success".format(result))

结果：

 get page 2s finished
 get page 3s finished
 in main:get page 3s success
 in main:get page 2s success
 get page 4s finished
 in main:get page 4s success

<6>wait()方法

wait方法可以让主线程阻塞，直到满足设定的要求。wait方法接收3个参数，等待的任务序列、超时时间以及等待条件。
等待条件return_when默认为ALL_COMPLETED,表明要等待所有的任务都借宿。可以看到运行结果中，确实是所有任务都完成了，主线程才打印出main，等待条件还可以设置为FIRST_COMPLETED,表示第一个任务完成就停止等待。

超时时间参数可以不设置：

wait()方法和as_completed(), map()没有关系。不管你是用as_completed()，还是用map()方法，你都可以在执行主线程之前使用wait()。
as_completed()和map()是二选一的。

# -*-coding:utf-8 -*-from concurrent.futures import ThreadPoolExecutor,wait,ALL_COMPLETED,FIRST_COMPLETEDimport time# 参数times用来模拟网络请求时间def get_html(times):    time.sleep(times)    print("get page {}s finished".format(times))    return times   # 创建线程池子# 设置最多2个线程运行，其他等待executor = ThreadPoolExecutor(max_workers=2)urls = [3,2,4]all_task = [executor.submit(get_html,(url)) for url in urls]wait(all_task,return_when=ALL_COMPLETED)print("main")# 结果# get page 2s finished# get page 3s finished# get page 4s finished# main

2、ProcessPoolExecutor多进程

<1>同步调用方式: 调用，然后等返回值，能解耦，但是速度慢

import datetimefrom concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutorfrom threading import current_threadimport time, random, osimport requestsdef task(name):    print('%s %s is running'%(name,os.getpid()))    #print(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))    if __name__ == '__main__':    p = ProcessPoolExecutor(4)  # 设置        for i in range(10):        # 同步调用方式，不仅要调用，还要等返回值        obj = p.submit(task, "进程pid：")  # 传参方式(任务名，参数),参数使用位置或者关键字参数        res = obj.result()    p.shutdown(wait=True)  # 关闭进程池的入口，等待池内任务运行结束    print("主")################################# 另一个同步调用的demodef get(url):    print('%s GET %s' % (os.getpid(),url))    time.sleep(3)    response = requests.get(url)    if response.status_code == 200:        res = response.text    else:        res = "下载失败"    return res  # 有返回值def parse(res):    time.sleep(1)    print("%s 解析结果为%s" %(os.getpid(),len(res)))if __name__ == "__main__":    urls = [        'https://www.baidu.com',        'Https://www.sina.com.cn',        'https://www.tmall.com',        'https://www.jd.com',        'https://www.python.org',        'https://www.openstack.org',        'https://www.baidu.com',        'https://www.baidu.com',        'https://www.baidu.com',    ]    p=ProcessPoolExecutor(9)    l=[]    start = time.time()    for url in urls:        future = p.submit(get,url)  # 需要等结果，所以是同步调用        l.append(future)        # 关闭进程池，等所有的进程执行完毕    p.shutdown(wait=True)    for future in l:        parse(future.result())    print('完成时间:',time.time()-start)    #完成时间: 13.209137678146362

<2>异步调用方式：只调用，不等返回值，可能存在耦合，但是速度快

def task(name):    print("%s %s is running" %(name,os.getpid()))    time.sleep(random.randint(1,3))if __name__ == '__main__':    p = ProcessPoolExecutor(4) # 设置进程池内进程    for i in range(10):        # 异步调用方式，只调用，不等返回值        p.submit(task,'进程pid：') # 传参方式(任务名,参数),参数使用位置参数或者关键字参数    p.shutdown(wait=True)  # 关闭进程池的入口，等待池内任务运行结束    print('主')##################################### 另一个异步调用的demodef get(url):    print('%s GET %s' % (os.getpid(),url))    time.sleep(3)    reponse = requests.get(url)    if reponse.status_code == 200:        res = reponse.text    else:        res = "下载失败"    parse(res)  # 没有返回值def parse(res):    time.sleep(1)    print('%s 解析结果为%s' %(os.getpid(),len(res)))if __name__ == '__main__':    urls = [        'https://www.baidu.com',        'https://www.sina.com.cn',        'https://www.tmall.com',        'https://www.jd.com',        'https://www.python.org',        'https://www.openstack.org',        'https://www.baidu.com',        'https://www.baidu.com',        'https://www.baidu.com',    ]    p = ProcessPoolExecutor(9)    start = time.time()    for url in urls:        future = p.submit(get,url)    p.shutdown(wait=True)    print("完成时间",time.time()-start)#  完成时间 6.293345212936401

<3>怎么使用异步调用方式，但同时避免耦合的问题？

(1)进程池：异步 + 回调函数，，cpu密集型，同时执行，每个进程有不同的解释器和内存空间，互不干扰

def get(url):    print('%s GET %s' % (os.getpid(), url))    time.sleep(3)    response = requests.get(url)    if response.status_code == 200:        res = response.text    else:        res = '下载失败'    return resdef parse(future):    time.sleep(1)    # 传入的是个对象，获取返回值 需要进行result操作    res = future.result()    print("res",)    print('%s 解析结果为%s' % (os.getpid(), len(res)))if __name__ == '__main__':    urls = [        'https://www.baidu.com',        'https://www.sina.com.cn',        'https://www.tmall.com',        'https://www.jd.com',        'https://www.python.org',        'https://www.openstack.org',        'https://www.baidu.com',        'https://www.baidu.com',        'https://www.baidu.com',    ]    p = ProcessPoolExecutor(9)    start = time.time()    for url in urls:        future = p.submit(get,url)        #模块内的回调函数方法，parse会使用future对象的返回值，对象返回值是执行任务的返回值        #回调应该是相当于parse(future)        future.add_done_callback(parse)   p.shutdown(wait=True)    print("完成时间",time.time()-start)#完成时间 33.79998469352722

(2)线程池：异步 + 回调函数，IO密集型主要使用方式,线程池:执行操作为谁有空谁执行

def get(url):    print("%s GET %s" %(current_thread().name,url))    time.sleep(3)    reponse = requests.get(url)    if reponse.status_code == 200:        res = reponse.text    else:        res = "下载失败"    return resdef parse(future):    time.sleep(1)    res = future.result()    print("%s 解析结果为%s" %(current_thread().name,len(res)))if __name__ == '__main__':    urls = [        'https://www.baidu.com',        'https://www.sina.com.cn',        'https://www.tmall.com',        'https://www.jd.com',        'https://www.python.org',        'https://www.openstack.org',        'https://www.baidu.com',        'https://www.baidu.com',        'https://www.baidu.com',    ]    p = ThreadPoolExecutor(4)    start = time.time()    for url in urls:        future = p.submit(get,url)        future.add_done_callback(parse)    p.shutdown(wait=True)    print("主",current_thread().name)    print("完成时间",time.time()-start)#完成时间 32.52604126930237

关于“python中ThreadPoolExecutor线程池和ProcessPoolExecutor进程池怎么使用”这篇文章的内容就介绍到这里，感谢各位的阅读！相信大家对“python中ThreadPoolExecutor线程池和ProcessPoolExecutor进程池怎么使用”知识都有一定的了解，大家如果还想学习更多知识，欢迎关注编程网Python频道。