铁乐学python_Day42_线程池
concurrent.futures 异步调用模块
- concurrent.futures模块提供了高度封装的异步调用接口
- ThreadPoolExecutor：线程池，提供异步调用
- ProcessPoolExecutor: 进程池，提供异步调用
基本方法
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)
回调函数
例： 进程池-异步调用
from concurrent.futures import ProcessPoolExecutor
import os, time, random
def task(n):
print('%s is runing' % os.getpid())
time.sleep(random.randint(1, 3))
return n ** 2
if __name__ == '__main__':
# 创建异步调用的进程池，最大工作进程为3
executor = ProcessPoolExecutor(max_workers=3)
futures = []
for i in range(11):
# submit提交任务
future = executor.submit(task, i)
futures.append(future)
# 关闭进程池及等待池内所有任务执行完毕后回收资源
executor.shutdown(True)
print('+++>')
for future in futures:
# result取得函数结果
print(future.result())
运行效果如下：
8984 is runing
8260 is runing
8772 is runing
8984 is runing
8772 is runing
8984 is runing
8260 is runing
8772 is runing
8772 is runing
8984 is runing
8772 is runing
+++>
0
1
4
9
16
25
36
49
64
81
100
例：线程池-异步调用-回调函数
import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
print(i*'*')
time.sleep(1)
return i
def callb(arg):
print(arg.result() ** 2)
if __name__ == '__main__':
thread_pool = ThreadPoolExecutor(5)
for i in range(1, 11):
# 回调函数，使用前面任务的函数结果做为传参到callb函数中异步执行
ret = thread_pool.submit(func, i).add_done_callback(callb)
thread_pool.shutdown()
运行效果如下：（这里并没有建列表，回调函数显示的结果是无序的）
*
**
***
****
*****
1
******
16
*******
9
********
25
*********
4
**********
36
49
81
64
100
例：线程池中map的用法（生成器）
from concurrent.futures import ThreadPoolExecutor
import time, random
def task(n):
print(n * '*')
time.sleep(random.randint(1, 3))
return n ** 2
if __name__ == '__main__':
executor = ThreadPoolExecutor(max_workers=3)
# for i in range(1,11):
# executor.submit(task,i)
ret = executor.map(task, range(1, 11))
# executor.map(task, range(1, 11))一行取代了上面注释掉的for+submit两行，
# 且这个时候只是运行了task函数，task函数return的结果并没有拿到，
# 如果要拿到它的结果，可以赋个变量名给它，它是一个生成器，for循环next取值显示就可以了。
for i in range(10):
print(next(ret))
运行效果如下：
*
**
***
****
1
*****
4
******
*******
********
9
*********
16
25
36
**********
49
64
81
100
例：回调函数
from concurrent.futures import ProcessPoolExecutor
import requests
import os
def get_page(url):
'''
get url请求是200被正常响应时的结果
:param url:
:return:
'''
print('<进程%s> get %s' % (os.getpid(), url))
respone = requests.get(url)
if respone.status_code == 200:
return {'url': url, 'text': respone.text}
def parse_page(res):
'''
将get_page函数的return结果url的值写入db文件中
:param res:
:return:
'''
res = res.result()
print('<进程%s> parse %s' % (os.getpid(), res['url']))
parse_res = 'url:<%s> size:[%s]\n' % (res['url'], len(res['text']))
with open('db.txt', 'a') as f:
f.write(parse_res)
if __name__ == '__main__':
urls = [
'https://www.baidu.com',
'https://www.python.org',
'https://www.openstack.org',
'https://help.github.com/',
'http://www.sina.com.cn/'
]
p = ProcessPoolExecutor(3)
for url in urls:
# 将前者get_page的return结果做为参数传入parse_page当中
# add 加，done 完成，callback 回调
p.submit(get_page, url).add_done_callback(parse_page)
# parse_page拿到的是一个future对象obj，需要用obj.result()拿到结果
运行效果如下：
<进程7264> get https://www.baidu.com
<进程8228> get https://www.python.org
<进程8988> get https://www.openstack.org
<进程7264> get https://help.github.com/
<进程3244> parse https://www.baidu.com
<进程7264> get http://www.sina.com.cn/
<进程3244> parse https://help.github.com/
<进程3244> parse http://www.sina.com.cn/
<进程3244> parse https://www.openstack.org
<进程3244> parse https://www.python.org
db.txt里内容：
url:<https://www.baidu.com> size:[2443]
url:<https://help.github.com/> size:[128491]
url:<http://www.sina.com.cn/> size:[584087]
url:<https://www.openstack.org> size:[63796]
url:<https://www.python.org> size:[48744]
浅析进程和线程的使用场景
- 当内存不需要共享，且高计算的时候，用进程；
- 当内存需要共享，且高IO的时候，用线程。
- 当并发很大的时候
- 多进程 ： 多个任务 —— 进程池 ：(数量)cpu个数、cpu个数+1
- 多线程 ：多个任务 —— 线程池 ：(数量)cpu个数*5
- 4CPU : 开5个进程 —— 每进程再开20条线程：可完成100个任务
- 例：50000并发: 5进程20线程500协程