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

一 .線程池(ThreadPoolExecutor)

https://www.cnblogs.com/nickchen121/p/11141751.html#autoid-3-0-0

#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函數的返回值  作為回調函數的參數  然后去執行回調函數