温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》
  • 首页 > 
  • 教程 > 
  • 开发技术 > 
  • python中ThreadPoolExecutor线程池和ProcessPoolExecutor进程池怎么使用

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

发布时间:2022-06-17 09:46:36 来源:亿速云 阅读:136 作者:iii 栏目:开发技术

这篇文章主要介绍了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 ThreadPoolExecutor
    import 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,没有完成返回False
    print( task1.done() )
    # cancel()方法用于取消某个任务,该任务没有放到线程池中才能被取消,如果已经放进线程池子中,则不能被取消
    # bool型,成功取消了返回True,没有取消返回False
    print( 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_completed
    import 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_completed
    import 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_COMPLETED
    import 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 datetime
    from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
    from threading import current_thread
    import time, random, os
    import requests
    def 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("主")
    ################
    ################
    # 另一个同步调用的demo
    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 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('主')
    ##################
    ##################
    # 另一个异步调用的demo
    def 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 res
    def 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 res
    def 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进程池怎么使用”知识都有一定的了解,大家如果还想学习更多知识,欢迎关注亿速云行业资讯频道。

    向AI问一下细节

    免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

    AI