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

以你之姓@ 2021-12-23 12:55 402阅读 0赞

# 使用concurrent.futures模块中的线程池与进程池 #

## 线程池与进程池 ##

以线程池举例，系统使用多线程方式运行时，会产生大量的线程创建与销毁，创建与销毁必定会带来一定的消耗，甚至导致系统资源的崩溃，这时使用线程池就是一个很好的解决方式。

“池”就说明了这里边维护了不止一个线程，线程池会提前创建好规定数量的线程，把需要使用多线程的任务提交给线程池，线程池会自己选择空闲的线程来执行提交的任务，任务完成后，线程并不会在池子中销毁，而是继续存在并等待完成下一个分配的任务。当线程池以满的时候，提交的线程会等待，也就是说线程池会有一个最大数量的运行线程限制。

进程池同样也是这个道理。

concurrent.futures模块为我们提供了ThreadPoolExecutor与ProcessPoolExecutor来使用线程进程池

## ThreadPoolExecutor ##

下面是一个简单的例子

from concurrent.futures import ThreadPoolExecutor
    import requests,time
    url_list = ['https://www.cnblogs.com/', 'https://www.csdn.net/', 'https://github.com/']
    def get_url(url):
        content = requests.get(url).content.decode()
        print(url+'已获取')
    
    pool = ThreadPoolExecutor(max_workers=3)
    
    start = time.time()
    for url in url_list:
        future = pool.submit(get_url,url)
        ＃　print(future)
    end = time.time()
    print(end-start)

输出的结果为：

> 0.0016434192657470703  
> [https://www.cnblogs.com/已获取][https_www.cnblogs.com]  
> [https://www.csdn.net/已获取][https_www.csdn.net]  
> [https://github.com/已获取][https_github.com]

例子中max\_workers为指定线程个数，pool.submit为提交任务到线程执行，get\_url为方法，url为参数

并且通过输出顺序可以看到线程池的执行并不会阻塞主线程的运行

print(future)被打了注释，现在我们取消注释运行一下：

> Future at 0x7ff6cfaa8860 state=running  
> Future at 0x7ff6ce965860 state=running  
> Future at 0x7ff6ce96e278 state=running  
> 0.006175518035888672  
> [https://www.cnblogs.com/已获取][https_www.cnblogs.com]  
> [https://www.csdn.net/已获取][https_www.csdn.net]  
> [https://github.com/已获取][https_github.com]

每提交一个任务后都会返回一个future对象，通过它可以查看任务运行的状态，state=running表示正在运行

future对象还有许多方法：

### future.done() ###

from concurrent.futures import ThreadPoolExecutor
    import requests,time
    url_list = ['https://www.cnblogs.com/', 'https://www.csdn.net/', 'https://github.com/']
    def get_url(url):
        content = requests.get(url).content.decode()
        print(url+'已获取')
    
    pool = ThreadPoolExecutor(max_workers=3)
    future_list = []
    start = time.time()
    for url in url_list:
        future = pool.submit(get_url,url)
        print(future.done())
        future_list.append(future)
    end = time.time()
    
    print(end-start)
    time.sleep(5)
    for future in future_list:
        print(future.done())

这里添加了future\_list,为了显示效果中间添加sleep,最后结果为：

> False  
> False  
> False  
> 0.001546621322631836  
> [https://www.cnblogs.com/已获取][https_www.cnblogs.com]  
> [https://www.csdn.net/已获取][https_www.csdn.net]  
> [https://github.com/已获取][https_github.com]  
> True  
> True  
> True

future.done()可以显示当前允许状态

### future.result() ###

from concurrent.futures import ThreadPoolExecutor
    import requests,time
    url_list = ['https://www.cnblogs.com/', 'https://www.csdn.net/', 'https://github.com/']
    def get_url(url):
        content = requests.get(url).content.decode()
        print(url+'已获取')
        return url
    
    pool = ThreadPoolExecutor(max_workers=3)
    future_list = []
    start = time.time()
    for url in url_list:
        future = pool.submit(get_url,url)
        print(future.result())
        future_list.append(future)
    end = time.time()
    
    print(end-start)
    for future in future_list:
        print(future.result())

结果为：

> [https://www.cnblogs.com/已获取][https_www.cnblogs.com]  
> [https://www.cnblogs.com/][https_www.cnblogs.com 1]  
> [https://www.csdn.net/已获取][https_www.csdn.net]  
> [https://www.csdn.net/][https_www.csdn.net 1]  
> [https://github.com/已获取][https_github.com]  
> [https://github.com/][https_github.com 1]  
> 2.0975613594055176  
> [https://www.cnblogs.com/][https_www.cnblogs.com 1]  
> [https://www.csdn.net/][https_www.csdn.net 1]  
> [https://github.com/][https_github.com 1]

可见result()方法可以得到任务的返回值，但会阻塞，因为不运行完怎么会得到返回值呢？

除此之外还有很多方法：

![1469015-20190325220526814-966878951.png][]

### 使用map方法 ###

与内建函数用法类似

### 使用wait方法 ###

from concurrent.futures import ThreadPoolExecutor,wait
    import requests,time
    url_list = ['https://www.cnblogs.com/', 'https://www.csdn.net/', 'https://github.com/']
    def get_url(url):
        content = requests.get(url).content.decode()
        print(url+'已获取')
        return url
    
    pool = ThreadPoolExecutor(max_workers=3)
    future_list = []
    start = time.time()
    for url in url_list:
        future = pool.submit(get_url,url)
        future_list.append(future)
    
    print(wait(future_list))
    end = time.time()
    print(end-start)

> [https://www.cnblogs.com/已获取][https_www.cnblogs.com]  
> [https://www.csdn.net/已获取][https_www.csdn.net]  
> [https://github.com/已获取][https_github.com]  
> DoneAndNotDoneFutures(done=\{Future at 0x7f7506447da0 state=finished returned str, Future at 0x7f75074c9828 state=finished returned str, Future at 0x7f75064477f0 state=finished returned str\}, not\_done=set())
> 
> 6.678021430969238

wait返回值是一个元组，元组里是已完成和未完成的两个集合，它的return\_when参数接受３个选项FIRST\_COMPLETED, FIRST\_EXCEPTION 和ALL\_COMPLETE，默认是ALL\_COMPLETE，意味着所有都完成，FIRST\_COMPLETED意味着有一个完成了就可以了， FIRST\_EXCEPTION是第一个出现异常就会停止wait

例如：

from concurrent.futures import ThreadPoolExecutor,wait
    import requests,time
    url_list = ['https://www.cnblogs.com/', 'https://www.csdn.net/', 'https://github.com/']
    def get_url(url):
        content = requests.get(url).content.decode()
        print(url+'已获取')
        return url
    
    def error(url):
        gg
    
    pool = ThreadPoolExecutor(max_workers=4)
    future_list = []
    start = time.time()
    future_list.append(pool.submit(error,'https://www.cnblogs.com/'))
    for url in url_list:
        future = pool.submit(get_url,url)
        future_list.append(future)
    
    print(wait(future_list,return_when='FIRST_EXCEPTION'))
    end = time.time()
    print(end-start)

> DoneAndNotDoneFutures(done=\{Future at 0x7fd1a5b95320 state=finished raised NameError\}, not\_done=\{Future at 0x7fd1a4b11a90 state=running, Future at 0x7fd1a4b11a20 state=running, Future at 0x7fd1a4c897f0 state=running\})  
> 0.001996755599975586  
> [https://www.cnblogs.com/已获取][https_www.cnblogs.com]  
> [https://www.csdn.net/已获取][https_www.csdn.net]  
> [https://github.com/已获取][https_github.com]

## ProcessPoolExecutor ##

进程池与线程池的使用方式基本相同，套用即可

转载于:https://www.cnblogs.com/sfencs-hcy/p/10597088.html

[https_www.cnblogs.com]: https://www.cnblogs.com/%E5%B7%B2%E8%8E%B7%E5%8F%96
[https_www.csdn.net]: https://www.csdn.net/%E5%B7%B2%E8%8E%B7%E5%8F%96
[https_github.com]: https://github.com/%E5%B7%B2%E8%8E%B7%E5%8F%96
[https_www.cnblogs.com 1]: https://www.cnblogs.com/
[https_www.csdn.net 1]: https://www.csdn.net/
[https_github.com 1]: https://github.com/
[1469015-20190325220526814-966878951.png]: /images/20211223/02433896dcd745939cea5e124325f3e6.png