# 池
# 進程池
# 線程池
# 為什么要有池？
# 10000
池
# 預先的開啟固定個數的進程數，當任務來臨的時候，直接提交給已經開好的進程
# 讓這個進程去執行就可以了
池的優點：
# 節省了進程，線程的開啟 關閉 切換都需要時間
# 並且減輕了操作系統調度的負擔
#
1、開啟進程池。任務少於進程數。開啟進程池，這里也要放到if name == 'main':下
import os,time,random
from concurrent.futures import ProcessPoolExecutor
def func():
print('start',os.getpid())
time.sleep(random.randint(1,3))
print('end',os.getpid())
if __name__ == '__main__':
p=ProcessPoolExecutor(5)
p.submit(func)
-----------結果：
start 4516
end 4516
2、開啟的進程最多和進程池的大小一樣。任務多於進程數
import os,time,random
from concurrent.futures import ProcessPoolExecutor
def func():
print('start',os.getpid())
time.sleep(random.randint(1,3))
print('end',os.getpid())
if __name__ == '__main__':
p=ProcessPoolExecutor(2)
for i in range(4):
p.submit(func)
----------------結果：
start 2548
start 5816
end 2548
start 2548
end 5816
start 5816
end 2548
end 5816
分析：2548執行完任務之后又來了一個任務再次執行了一遍，總共開啟的進程最多是定義的進程池的大小。這里是以2個進程最大效率的執行這提交的4個任務
3、池是實現異步的
import os,time,random
from concurrent.futures import ProcessPoolExecutor
def func():
print('start',os.getpid())
time.sleep(random.randint(1,3))
print('end',os.getpid())
if name == 'main':
p=ProcessPoolExecutor(2)
for i in range(4):
p.submit(func)
print('main',os.getpid())
-------------結果：
main 5512
start 5352
start 5376
end 5376
start 5376
end 5352
start 5352
end 5376
end 5352
先打印的的main。我們常常需要的是等提交到進程池中的任務都執行完畢后再往下執行主進程中的代碼。這時需要阻塞，池的阻塞是池.shutdown()
4、submit和shutdown配合，進程池中任務都執行完畢再(做某件事)往下執行主進程代碼(shutdown后面的)。池.shutdown()
import os,time,random
from concurrent.futures import ProcessPoolExecutor
def func():
print('start',os.getpid())
time.sleep(random.randint(1,3))
print('end',os.getpid())
if name == 'main':
p=ProcessPoolExecutor(2)
for i in range(4):
p.submit(func)
p.shutdown() #關閉池之后就不能繼續提交任務，並且會阻塞，直到已經提交的任務完成
print('main',os.getpid())
------------------結果:
start 5780
start 796
end 5780
start 5780
end 5780
start 5780
end 796
end 5780
main 5640
關閉池之后繼續提交任務報錯：RuntimeError: cannot schedule new futures after shutdown
5、池執行的函數中傳參數，submit后面直接傳參，但是池函數必須是第一個參數。
from concurrent.futures import ProcessPoolExecutor
def func(i):
print(i)
if name == 'main':
p=ProcessPoolExecutor(2)
for i in range(4):
p.submit(func,i)
p.shutdown()
----------結果：
0
1
2
3
源碼：def submit(self, fn, *args, **kwargs):
6、
import os,time,random
from concurrent.futures import ProcessPoolExecutor
def func(i):
time.sleep(random.randint(1,3))
return i*i
if name == 'main':
p=ProcessPoolExecutor(2)
for i in range(4):
ret=p.submit(func,i)
print(ret,ret.result()) #這里有阻塞，取返回值結果
p.shutdown()
print('main',os.getpid())
-------------結果：
<Future at 0x2913090 state=finished returned int> 0
<Future at 0x293b770 state=finished returned int> 1
<Future at 0x293b810 state=finished returned int> 4
<Future at 0x293b7f0 state=finished returned int> 9
main 5384
注釋：ret后面的返回值是個Future對象，這個對象.result()返回池中任務函數的返回值。這個返回值在for循環中打印，那么執行完一個任務打印出結果才會提交下一個任務，這樣就不是異步了，而是同步了，不是我們用池實現異步的效果。
7、既能實現並發又能獲得返回值。獲得返回值的時候不在提交任務的循環中，獲得一個對象就放到列表中，從列表中取返回值，不影響它任務的提交。
import os,time,random
from concurrent.futures import ProcessPoolExecutor
def func(i):
time.sleep(random.randint(1,3))
return i*i
if name == 'main':
p=ProcessPoolExecutor(2)
li=[]
for i in range(4):
ret=p.submit(func,i)
li.append(ret)
for ret in li:print('ret-->',ret.result()) #ret這里是同步阻塞
p.shutdown()
print('main',os.getpid())
--------------結果：
ret--> 0
ret--> 1
ret--> 4
ret--> 9
main 5928
8、進程池的特點：
開銷大
一個池中的任務個數限制了我們程序的並發個數
9、線程池提交一個任務加傳參
從並發.未來導入線程程池執行者，實例化線程池執行者並指定線程個數，對象.提交(任務函數，參數（多個參數都可以）)
import os,time,random
from concurrent.futures import ThreadPoolExecutor
def func(i):
print('start', os.getpid(),' arg:',i)
time.sleep(random.randint(1,3))
tp=ThreadPoolExecutor(20) #線程個數能開很多，一般開啟cpu個數乘以4或5
tp.submit(func,1)
----------------結果：
start 2844 arg: 1
10、提交多個線程池任務
for 循環提交多個任務
import os,time,random
from concurrent.futures import ThreadPoolExecutor
def func(i):
print('start', os.getpid(),' arg:',i)
time.sleep(random.randint(1,3))
tp=ThreadPoolExecutor(2)
for i in range(5):
tp.submit(func,i)
------------結果：
start 2100 arg: 0
start 2100 arg: 1
start 2100 arg: 2
start 2100 arg: 3
start 2100 arg: 4
這個創建的順序和執行的順序好像差不多，但是是實現並發的。
11、線程池獲取線程任務函數的返回值
接收提交，追加到列表，循環列表中對象,對象.結果()
import os,time,random
from concurrent.futures import ThreadPoolExecutor
def func(i):
print('start', os.getpid(),' arg:',i)
time.sleep(random.randint(1,3))
return i**2
tp=ThreadPoolExecutor(2)
ret_li=[]
for i in range(5):
ret=tp.submit(func,i)
ret_li.append(ret)
for ret in ret_li:print('ret-->',ret.result())
----------------結果：
start 2856 arg: 0
start 2856 arg: 1
start 2856 arg: 2
ret--> 0
start 2856 arg: 3
ret--> 1
start 2856 arg: 4
ret--> 4
ret--> 9
ret--> 16
12、線程池中任務都結束才能再執行的代碼
tp.shutdown()線程池對象.關閉()，池中任務都結束才能執行關閉之后的代碼
import os,time,random
from concurrent.futures import ThreadPoolExecutor
def func(i):
print('start', os.getpid(),' arg:',i)
time.sleep(random.randint(1,3))
tp=ThreadPoolExecutor(2)
for i in range(4):
ret=tp.submit(func,i)
tp.shutdown()
print('main')
------------------結果：
start 1468 arg: 0
start 1468 arg: 1
start 1468 arg: 2
start 1468 arg: 3
main
13、線程池其它方法map（省代碼量批量提交任務）
批量創建線程池任務和創建好后獲取每個任務函數返回值更簡便
任務返回值列表變量=線程池對象.map(任務函數，傳參可迭代對象)。創建任務數是可迭代對象個數
import os,time,random
from concurrent.futures import ThreadPoolExecutor
def func(i):
print('start', os.getpid(),' arg:',i)
time.sleep(random.randint(1,3))
return i**2
tp=ThreadPoolExecutor(2)
ret=tp.map(func,range(5))
for i in ret:print(i)
---------------結果：
start 2976 arg: 0
start 2976 arg: 1
start 2976 arg: 2
start 2976 arg: 3
0
1
start 2976 arg: 4
4
print(ret,type(ret))
--------結果：
<generator object Executor.map.
9
16
14、多線程爬取網頁，獲取到結果后使用回調函數分析網頁
import requests
from concurrent.futures import ThreadPoolExecutor
def get_page(url):
res = requests.get(url)
return {'url': url, 'content': res.text} #ret的結果會作為參數返回給綁定的函數
def parserpage(ret):
dic = ret.result()
print(dic['url'])
tp = ThreadPoolExecutor(5)
url_lst = [
'http://www.baidu.com', # 3
'http://www.cnblogs.com', # 1
'http://www.douban.com', # 1
'http://www.tencent.com',
'http://www.cnblogs.com/Eva-J/articles/8306047.html',
'http://www.cnblogs.com/Eva-J/articles/7206498.html',
]
＃方法一：回調函數，使用回調函數簡便而且效率高
for url in url_lst:
ret = tp.submit(get_page, url)
ret.add_done_callback(parserpage) #綁定的函數中的內容#誰先執行完誰先使用回調函數
-------------------結果：
http://www.baidu.com
http://www.cnblogs.com/Eva-J/articles/7206498.html
http://www.tencent.com
http://www.cnblogs.com/Eva-J/articles/8306047.html
http://www.douban.com
http://www.cnblogs.com
＃方法二：非回調函數。
ret_l = []
for url in url_lst:
ret = tp.submit(get_page, url)
ret_l.append(ret)
for ret in ret_l:
parserpage(ret)
print("ret_l:", ret_l)
--------------------結果：
http://www.baidu.com
http://www.cnblogs.com
http://www.douban.com
http://www.tencent.com
http://www.cnblogs.com/Eva-J/articles/8306047.html
http://www.cnblogs.com/Eva-J/articles/7206498.html
ret_l: [<Future at 0x2db7150 state=finished returned dict>, <Future at 0x2e152d0 state=finished re
15、回調函數的使用
import time
from concurrent.futures import ThreadPoolExecutor
def son():
print(123)
time.sleep(3)
return 123
def call_back(num):
print(num.result())
t = ThreadPoolExecutor(20)
obj = t.submit(son)
print('main ： ',obj)
obj.add_done_callback(call_back)
----------結果；
123
main ： <Future at 0x39f5b0 state=running>
123
代碼分析：
導入類：從並發.未來導入線程池執行者
創建池對象：線程池(大小)
獲取返回值：用變量接收池對象.提交（任務函數），即未來對象
執行回調函數：未來對象.添加完成回調(回調函數)
回調函數接收一個參數，這個參數是obj未來對象，回調函數中取任務函數返回值就是這個參數.result()
16、線程池小結