一、什么是進程池或線程池

池:
池的是為了限制進程數和線程數
什么是時候該用進程池/線程池:
當python程序是 計算密集型 且并發的任務量遠大于計算機所能承受的范圍,
無法一次性開啟過多的任務數量就應該考慮使用進程池和線程池

二、理解同步、異步、

同步: 提交了一個任務,必須等任務執行完了,才能執行下一個.無論該任務是否存在阻塞,同步調用都會原地等待,相當于串行
異步: 提交了一個任務,可以不用等任務的執行結果,直接執行下一個,并發的去執行

三、multiprocess.Pool模塊實現進程池

Pool([numprocess [,initializer [, initargs]]]) ：創建進程池

1. numprocess：要創建的進程數，如果省略，將默認使用 cpu_count() 的值
2. initializer：是每個工作進程啟動時要執行的可調用對象，默認為None
3. initargs：是要傳給initializer的參數組

3.1.1 Pool使用方法

p.apply(func [, args [, kwargs]]) ： 在一個池工作進程中執行func(*args,**kwargs),然后返回結果。需要強調的是： 此操作并不會在所有池工作進程中并執行func函數。如果要通過不同參數并發地執行func函數，必須從不同線程調用p.apply()函數或者使用p.apply_async()

p.apply_async(func [, args [, kwargs]]) ： 在一個池工作進程中執行func(*args,**kwargs),然后返回結果。此方法的結果是AsyncResult類的實例，callback是可調用對象，接收輸入參數。當func的結果變為可用時，將理解傳遞給callback。callback禁止執行任何阻塞操作，否則將接收其他異步操作中的結果。

p.close() ：關閉進程池，防止進一步操作。如果所有操作持續掛起，它們將在工作進程終止前完成

P.join() ：等待所有工作進程退出。此方法只能在 close() 或 teminate() 之后調用

obj.get() ：返回結果，如果有必要則等待結果到達。timeout是可選的。如果在指定時間內還沒有到達，將引發一場。如果遠程操作中引發了異常，它將在調用此方法時再次被引發。

obj.ready() ：如果調用完成，返回True

obj.successful() ：如果調用完成且沒有引發異常，返回True，如果在結果就緒之前調用此方法，引發異常

obj.wait([timeout]) ：等待結果變為可用。

obj.terminate() ：立即終止所有工作進程，同時不執行任何清理或結束任何掛起工作。如果p被垃圾回收，將自動調用此函數

3.1.1 代碼實例——multiprocess.Pool

同步運行方式(串行等待)：

          
            def task(i):
    print(f'進程 {current_process().name} 在執行任務 {i}')
    time.sleep(1)
    return i**2,current_process().name

### Pool方式
p = Pool(3) # 創建進程池對象,可容納3個進程

### 同步運行(串行等待)
for i in range(20):
    res = p.apply(task, args=(i,))
    print(res)
          
        

異步運行方式：

          
            def task(i):
    print(f'進程 {current_process().name} 在執行任務 {i}')
    time.sleep(1)
    return i**2,current_process().name

for i in range(20):
    res = p.apply_async(task, args=(i,), callback=lambda res: print(res))    # 異步處理并設置了
    
    p.close()   # 先關閉進程池
    p.join()    # 在等待進程結果
          
        

四、Python標準模塊——concurrent.futures實現進程池和線程池

4.1 介紹

concurrent.futures模塊提供了高度封裝的異步調用接口

ThreadPoolExecutor：線程池，提供異步調用

ProcessPoolExecutor：進程池，提供異步調用

4.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() ：取消某個任務

4.3 代碼實例——ProcessPoolExecutor

方式1:

異步的執行任務,在把每次返回的對象添加到列表中,最后統一得到結果

          
            from concurrent.futures import ProcessPoolExecutor
from multiprocessing import current_process,Pool
import time

def task(i):
    print(f'進程 {current_process().name} 在執行任務 {i}')
    time.sleep(1)
    return i**2,current_process().name

def parse(future):
    # 處理拿到的結果
    print(future.result())  # 在這里拿到結果
    
if __name__ == '__main__':
    ### ProcessPoolExecutor方式(異步調用接口)
    pool = ProcessPoolExecutor(3)   # maxsize 填最大數
    
    ### ProcessPoolExecutor方式1:
    futures = []
    for i in range(20):
        future = pool.submit(task, i)   # 提交異步任務
        futures.append(future)
    pool.shutdown() # 這句話相當于 進程池關閉和進程的join方法
    for future in futures:
        print(future.result())
          
        

方式2:

每執行一個進程都拿到結果.等于把異步任務變成同步任務,原地等待結果

          
            from concurrent.futures import ProcessPoolExecutor
from multiprocessing import current_process,Pool
import time

def task(i):
    print(f'進程 {current_process().name} 在執行任務 {i}')
    time.sleep(1)
    return i**2,current_process().name

def parse(future):
    # 處理拿到的結果
    print(future.result())  # 在這里拿到結果
    
if __name__ == '__main__':
    ### ProcessPoolExecutor方式(異步調用接口)
    pool = ProcessPoolExecutor(3)   # maxsize 填最大數
    
### ProcessPoolExecutor方式2:  
    for i in range(20):
        future = pool.submit(task, i)   # 提交異步任務
        print(future.result())      # 如果加了這一句就變成 同步 串行了,獲取每一次任務的執行結果
    pool.shutdown() # 這句話相當于 進程池關閉和進程的join方法
          
        

方式3

通過回調函數來異步的執行任務,并得到每一次的結果

          
            from concurrent.futures import ProcessPoolExecutor
from multiprocessing import current_process,Pool
import time

def task(i):
    print(f'進程 {current_process().name} 在執行任務 {i}')
    time.sleep(1)
    return i**2,current_process().name

def parse(future):
    # 處理拿到的結果
    print(future.result())  # 在這里拿到結果
    
if __name__ == '__main__':
    ### ProcessPoolExecutor方式(異步調用接口)
    pool = ProcessPoolExecutor(3)   # maxsize 填最大數
    
##### ProcessPoolExecutor方式3:  
    for i in range(20):
        future = pool.submit(task, i)  # 提交異步任務
        future.add_done_callback(parse)   # 因此不能直接寫result,要通過回調函數
    pool.shutdown() # 這句話相當于 進程池關閉和進程的join方法
          
        

4.4 代碼實例——ThreadPoolExecutor

方式1：

異步的執行任務,在把每次返回的對象添加到列表中,最后統一得到結果

          
            from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
import time

def task(i):
    print(f'{currentThread().name} 在執行任務 {i}')
    time.sleep(1)
    return i**2

def parse(future):
    # 處理拿到的結果
    print(future.result())
    
if __name__ == '__main__':
    ### ThreadPoolExecutor(異步調用接口)
    pool = ThreadPoolExecutor(3)  # 池子里只有3個線程

    ### ProcessPoolExecutor方式1:  
    futures = []
    for i in range(20):
        future = pool.submit(task, i)   # 提交異步任務
        futures.append(future)
    pool.shutdown() # 這句話相當于 線程池關閉和線程的join方法
    for future in futures:
        print(future.result())
          
        

方式2：

每執行一個線程都拿到結果.等于把異步任務變成同步任務,原地等待結果了

          
            from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
import time

def task(i):
    print(f'{currentThread().name} 在執行任務 {i}')
    time.sleep(1)
    return i**2

def parse(future):
    # 處理拿到的結果
    print(future.result())
    
if __name__ == '__main__':
    ### ThreadPoolExecutor(異步調用接口)
    pool = ThreadPoolExecutor(3)  # 池子里只有3個線程
    
    ### ProcessPoolExecutor方式2:  
    for i in range(20):
        future = pool.submit(task, i)   # 提交異步任務
        print(future.result())      # 如果加了這一句就變成 同步 串行了,獲取每一次任務的執行結果
    pool.shutdown() # 這句話相當于 線程池關閉和線程的join方法
          
        

方式3：

          
            from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
import time

def task(i):
    print(f'{currentThread().name} 在執行任務 {i}')
    time.sleep(1)
    return i**2

def parse(future):
    # 處理拿到的結果
    print(future.result())
    
if __name__ == '__main__':
    ### ThreadPoolExecutor(異步調用接口)
    pool = ThreadPoolExecutor(3)  # 池子里只有3個線程
    
    ##### ProcessPoolExecutor方式3:  通過回調函數來異步的執行任務,并得到每一次的結果
    for i in range(20):
        future = pool.submit(task, i)  # 提交異步任務
        future.add_done_callback(parse)   # 因此不能直接寫result,要通過回調函數
    pool.shutdown() # 這句話相當于 線程池關閉和線程的join方法