《Python》線程池、攜程
一、線程池(concurrent.futures模塊)
#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) 回調函數
import time from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutordef func(i): print(‘thread‘, i) time.sleep(1) print(‘thread %s end‘ % i) # 線程池,提供異步調用 tp = ThreadPoolExecutor(5) tp.submit(func, 1) tp.shutdown() print(‘主線程‘) # 進程池,提供異步調用 if __name__ == ‘__main__‘: tp = ProcessPoolExecutor(5) tp.submit(func, 1) tp.shutdown() print(‘主線程‘)
import time from concurrent.futures import ThreadPoolExecutor from threading import current_thread def func(i): print(‘thread‘, i, current_thread().ident) time.sleep(1) print(‘thread %s end‘ % i) tp = ThreadPoolExecutor(5) for i in range(20): tp.submit(func, i) tp.shutdown() print(‘主線程‘) # 獲取返回值 import time from concurrent.futures import ThreadPoolExecutor from threading import current_thread def func(i): print(‘thread‘, i, current_thread().ident) time.sleep(1) print(‘thread %s end‘ % i) return i * ‘*‘ tp = ThreadPoolExecutor(5) ret_l = [] for i in range(20): ret = tp.submit(func, i) ret_l.append(ret) for ret in ret_l: print(ret.result()) print(‘主線程‘)
# map的用法 import time from concurrent.futures import ThreadPoolExecutor def func(i): print(‘thread‘, i) time.sleep(1) print(‘thread %s end‘ % i) return i * ‘*‘ tp = ThreadPoolExecutor(5) res = tp.map(func, range(20)) for i in res: print(i)
# 回調函數 import time import os from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor from threading import current_thread def func(i): print(‘thread‘, i, current_thread().ident, os.getpid()) time.sleep(1) print(‘thread %s end‘ % i) return i * ‘*‘ def call_back(arg): print(‘call back : ‘, current_thread().ident, os.getpid()) print(‘ret : ‘, arg.result())
# 線程池的回調函數是由子線程完成的 tp = ThreadPoolExecutor(5) ret_l = [] for i in range(20): tp.submit(func, i).add_done_callback(call_back) print(‘主線程‘, current_thread().ident, os.getpid())
# 進程池的回調函數是由主進程完成的 if __name__ == ‘__main__‘: tp = ProcessPoolExecutor(5) ret_l = [] for i in range(20): tp.submit(func, i).add_done_callback(call_back) print(‘主線程‘, current_thread().ident, os.getpid())
二、協程
之前我們學習了線程、進程的概念,了解了在操作系統中進程是資源分配的最小單位,線程是CPU調度的最小單位。按道理來說我們已經算是把cpu的利用率提高很多了。但是我們知道無論是創建多進程還是創建多線程來解決問題,都要消耗一定的時間來創建進程、創建線程、以及管理他們之間的切換。
隨著我們對於效率的追求不斷提高,基於單線程來實現並發又成為一個新的課題,即只用一個主線程(很明顯可利用的cpu只有一個)情況下實現並發。這樣就可以節省創建線進程所消耗的時間。
為此我們需要先回顧下並發的本質:切換+保存狀態
cpu正在運行一個任務,會在兩種情況下切走去執行其他的任務(切換由操作系統強制控制),一種情況是該任務發生了阻塞,另外一種情況是該任務計算的時間過長
ps:在介紹進程理論時,提及進程的三種執行狀態,而線程才是執行單位,所以也可以將上圖理解為線程的三種狀態
一:其中第二種情況並不能提升效率,只是為了讓cpu能夠雨露均沾,實現看起來所有任務都被“同時”執行的效果,如果多個任務都是純計算的,這種切換反而會降低效率。
為此我們可以基於yield來驗證。yield本身就是一種在單線程下可以保存任務運行狀態的方法,我們來簡單復習一下:
# 1. yiled可以保存狀態,yield的狀態保存與操作系統的保存線程狀態很像,但是yield是代碼級別控制的,更輕量級 # 2. send可以把一個函數的結果傳給另外一個函數,以此實現單線程內程序之間的切換
![技術分享圖片](/img/jia.gif)
# 串行執行 import time def consumer(res): ‘‘‘任務1:接收數據,處理數據‘‘‘ pass def producer(): ‘‘‘任務2:生產數據‘‘‘ res=[] for i in range(10000000): res.append(i) return res start=time.time() # 串行執行 res=producer() consumer(res) # 寫成consumer(producer())會降低執行效率 stop=time.time() print(stop-start) # 1.5536692142486572 #基於yield並發執行 import time def consumer(): ‘‘‘任務1:接收數據,處理數據‘‘‘ while True: x=yield def producer(): ‘‘‘任務2:生產數據‘‘‘ g=consumer() next(g) for i in range(10000000): g.send(i) start=time.time() # 基於yield保存狀態,實現兩個任務直接來回切換,即並發的效果 # PS:如果每個任務中都加上打印,那麽明顯地看到兩個任務的打印是你一次我一次,即並發執行的. producer() stop=time.time() print(stop-start) #2.0272178649902344單純地切換反而會降低運行效率
二:第一種情況的切換。在任務一遇到io情況下,切到任務二去執行,這樣就可以利用任務一阻塞的時間完成任務二的計算,效率的提升就在於此。
![技術分享圖片](/img/jia.gif)
import time def consumer(): ‘‘‘任務1:接收數據,處理數據‘‘‘ while True: x=yield def producer(): ‘‘‘任務2:生產數據‘‘‘ g=consumer() next(g) for i in range(10000000): g.send(i) time.sleep(2) start=time.time() producer() # 並發執行,但是任務producer遇到io就會阻塞住,並不會切到該線程內的其他任務去執行 stop=time.time() print(stop-start)yield無法做到遇到io阻塞
對於單線程下,我們不可避免程序中出現io操作,但如果我們能在自己的程序中(即用戶程序級別,而非操作系統級別)控制單線程下的多個任務能在一個任務遇到io阻塞時就切換到另外一個任務去計算,這樣就保證了該線程能夠最大限度地處於就緒態,即隨時都可以被cpu執行的狀態,相當於我們在用戶程序級別將自己的io操作最大限度地隱藏起來,從而可以迷惑操作系統,讓其看到:該線程好像是一直在計算,io比較少,從而更多的將cpu的執行權限分配給我們的線程。
協程的本質就是在單線程下,由用戶自己控制一個任務遇到io阻塞了就切換另外一個任務去執行,以此來提升效率。為了實現它,我們需要找尋一種可以同時滿足以下條件的解決方案:
# 1. 可以控制多個任務之間的切換,切換之前將任務的狀態保存下來,以便重新運行時,可以基於暫停的位置繼續執行。 # 2. 作為1的補充:可以檢測io操作,在遇到io操作的情況下才發生切換
協程介紹:
協程:是單線程下的並發,又稱微線程,纖程。英文名Coroutine。一句話說明什麽是線程:協程是一種用戶態的輕量級線程,即協程是由用戶程序自己控制調度的。
需要強調的是:
#1. python的線程屬於內核級別的,即由操作系統控制調度(如單線程遇到io或執行時間過長就會被迫交出cpu執行權限,切換其他線程運行) #2. 單線程內開啟協程,一旦遇到io,就會從應用程序級別(而非操作系統)控制切換,以此來提升效率(!!!非io操作的切換與效率無關)
對比操作系統控制線程的切換,用戶在單線程內控制協程的切換
優點如下:
#1. 協程的切換開銷更小,屬於程序級別的切換,操作系統完全感知不到,因而更加輕量級 #2. 單線程內就可以實現並發的效果,最大限度地利用cpu
缺點如下:
#1. 協程的本質是單線程下,無法利用多核,可以是一個程序開啟多個進程,每個進程內開啟多個線程,每個線程內開啟協程 #2. 協程指的是單個線程,因而一旦協程出現阻塞,將會阻塞整個線程
總結協程特點:
- 必須在只有一個單線程裏實現並發
- 修改共享數據不需加鎖
- 用戶程序裏自己保存多個控制流的上下文棧
- 附加:一個協程遇到IO操作自動切換到其它協程(如何實現檢測IO,yield、greenlet都無法實現,就用到了gevent模塊(select機制))
Greenlet 模塊
安裝 :pip3 install greenlet
# greenlet實現狀態切換 import time from greenlet import greenlet def eat(): print(‘eating 1‘) g2.switch() time.sleep(1) print(‘eating 2‘) def play(): print(‘playing 1‘) time.sleep(1) print(‘playing 2‘) g1.switch() g1 = greenlet(eat) g2 = greenlet(play) g1.switch()
單純的切換(在沒有io的情況下或者沒有重復開辟內存空間的操作),反而會降低程序的執行速度
![技術分享圖片](/img/jia.gif)
# 順序執行 import time def f1(): res=1 for i in range(100000000): res+=i def f2(): res=1 for i in range(100000000): res*=i start=time.time() f1() f2() stop=time.time() print(‘run time is %s‘ %(stop-start)) #10.985628366470337 # 切換 from greenlet import greenlet import time def f1(): res=1 for i in range(100000000): res+=i g2.switch() def f2(): res=1 for i in range(100000000): res*=i g1.switch() start=time.time() g1=greenlet(f1) g2=greenlet(f2) g1.switch() stop=time.time() print(‘run time is %s‘ %(stop-start)) # 52.763017892837524效率對比
greenlet只是提供了一種比generator更加便捷的切換方式,當切到一個任務執行時如果遇到io,那就原地阻塞,仍然是沒有解決遇到IO自動切換來提升效率的問題。
單線程裏的這20個任務的代碼通常會既有計算操作又有阻塞操作,我們完全可以在執行任務1時遇到阻塞,就利用阻塞的時間去執行任務2。。。。如此,才能提高效率,這就用到了Gevent模塊。
Gevent 模塊
安裝:pip3 install gevent
Gevent 是一個第三方庫,可以輕松通過gevent實現並發同步或異步編程,在gevent中用到的主要模式是Greenlet, 它是以C擴展模塊形式接入Python的輕量級協程。 Greenlet全部運行在主程序操作系統進程的內部,但它們被協作式地調度。
g1=gevent.spawn(func,1,,2,3,x=4,y=5) # 創建一個協程對象g1,spawn括號內第一個參數是函數名,如eat,後面可以有多個參數,可以是位置實參或關鍵字實參,都是傳給函數eat的 g2=gevent.spawn(func2) g1.join() #等待g1結束 g2.join() #等待g2結束 #或者上述兩步合作一步:gevent.joinall([g1,g2]) g1.value#拿到func1的返回值
《Python》線程池、攜程