七.併發編程 (線程池,返回值，回調函數)

一 .線程池(ThreadPoolExecutor)

http://www.javashuo.com/article/p-dciirvhe-cz.html

#1 介紹
concurrent.futures模塊提供了高度封裝的異步調用接口
ThreadPoolExecutor：線程池，提供異步調用
ProcessPoolExecutor: 進程池，提供異步調用
Both implement the same interface, which is defined by the abstract Executor class.

#2 基本方法
#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)
回調函數

# done()
判斷某一個線程是否完成

# cancle()
取消某個任務

1. 線程池異步

import  time
from concurrent.futures import ThreadPoolExecutor
def fun(n):
    time.sleep(1)
    print("用線程池啓動的個數爲%s"%n)
po=ThreadPoolExecutor(max_workers=5) # 默認不超過cpu個數*5
for i in  range(5):
     po.submit(fun,i)   # 線程池是用submit來提交 傳遞參數 異步提交任務

print("主線程")

主線程
用線程池啓動的個數爲3
用線程池啓動的個數爲0
用線程池啓動的個數爲2
用線程池啓動的個數爲1
用線程池啓動的個數爲4
進程已結束，退出代碼 0

import  time

from concurrent.futures import ThreadPoolExecutor

def fun(n):
    time.sleep(1)
    print("啓動線程的個數爲%s"%n)
po=ThreadPoolExecutor(max_workers=5) #  默認不超過cpu個數*5
for i in  range(6):
     po.submit(fun,i)     #  線程池是用submit來提交  傳遞參數  異步提交任務
po.shutdown()    # 這裏裏面自帶了close join
print("主線程")

啓動線程的個數爲0
啓動線程的個數爲4
啓動線程的個數爲2
啓動線程的個數爲3
啓動線程的個數爲1
啓動線程的個數爲5
主線程

進程已結束，退出代碼 0

 2.線程返回值

import  time
from concurrent.futures import ThreadPoolExecutor

def fun(n):
    time.sleep(1)
    return  n*n

po=ThreadPoolExecutor(max_workers=5) #  默認不超過cpu個數*5
ll=[]
for i in  range(6):
     po.submit(fun,i)     #  線程池是用submit來提交  傳遞參數  異步提交任務
     ll.append(po)
po.shutdown()  #  裏面自帶close json
print("主線程")
for p in ll:print(p.result())   #     result() 來獲取返回值

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

import os,time,random
def task(n):
    print('%s線程的id號' %os.getpid())
    time.sleep(random.randint(1,3))
    return n**2

if __name__ == '__main__':
    executor=ProcessPoolExecutor(max_workers=3)
    futures=[]
    for i in range(5):
        future=executor.submit(task,i)
        futures.append(future)
    executor.shutdown(True)

    print('主線程!!!!!!!!!!!!!!!!!!11')
    for future in futures:
        print(future.result())

3256線程的id號
1228線程的id號
7608線程的id號
7608線程的id號
3256線程的id號
主線程!!!!!!!!!!!!!!!!!!11
0
1
4
9
16

進程已結束，退出代碼 0

map是獲取不到返回值

import  time
from concurrent.futures import ThreadPoolExecutor
def fun(n):
    time.sleep(1)
    return  n*n

po=ThreadPoolExecutor(max_workers=5) #  默認不超過cpu個數*5
po.map(fun,range(5))  #  map是拿不到返回值的
print("主線程")

from concurrent.futures import ThreadPoolExecutor,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__':

    executor=ThreadPoolExecutor(max_workers=3)

    # for i in range(11):
    #     future=executor.submit(task,i)

    executor.map(task,range(1,12)) #map取代了for+submit
    
4468 is runing
4468 is runing
4468 is runing
4468 is runing
4468 is runing
4468 is runing
4468 is runing
4468 is runing
4468 is runing
4468 is runing
4468 is runing
進程已結束，退出代碼 0

 

3. 線程回調函數(和進程使用方式同樣)

import  time,os
from concurrent.futures import ThreadPoolExecutor

def fun(n):
    time.sleep(1)
    return  n*5

def show(a):
    print(a)
    print(a.result())  # 獲取結果

    print("線程號",os.getpid())   

po=ThreadPoolExecutor(max_workers=5)

for i in  range(2):
     po.submit(fun,5).add_done_callback(show)
print("主線程",os.getpid())

主進程 9808
<Future at 0x2466fe4b5c0 state=finished returned int>
25
線程號 9808
<Future at 0x2466fb42b00 state=finished returned int>
25
線程號 9808

進程已結束，退出代碼 0

 

 

2	須要回調函數的場景：進程池中任何一個任務一旦處理完了，就當即告知主進程： 我好了額，你能夠處理個人結果了。主進程則調用一個函數去處理該結果，該函數即回調函數

# 先執行fun1函數 fun1函數的返回值  做爲回調函數的參數  而後去執行回調函數