参考网址
www.cnblogs.com/venvive/p/1…
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
3/总结
1、线程不是越多越好,会涉及cpu上下文的切换(会把上一次的记载保存)。
2、进程比线程消耗资源,进程相当于一个工厂,工厂里有很多人,里边的人一起享受着福利资源,,一个进程里默许只要一个主线程,比方:敞开程序是进程,里边履行的是线程,线程仅仅一个进程创立多个人一起去作业。
3、线程里有GIL大局解锁器:不允许cpu调度
4、核算密度型适用于多进程
5、线程:线程是核算机中作业的最小单元
6、进程:默许有主线程 (帮作业)能够多线程共存
7、协程:一个线程,一个进程做多个使命,运用进程中一个线程去做多个使命,微线程
8、GIL大局解释器锁:确保同一时间只要一个线程被cpu调度