制實(shí)際應(yīng)用場景舉例說明)
前言線程共享內(nèi)存所以只要兩個(gè)線程碰同一個(gè)可變對象就必須想清楚「怎么同步」。但同步不等于「到處加鎖」——threading模塊提供了五六種原語每種解決的是不同形狀的協(xié)調(diào)問題有的保護(hù)一段臨界區(qū)有的負(fù)責(zé)線程之間傳消息有的只是發(fā)個(gè)信號還有的是限制同時(shí)干活的人數(shù)。新手最常見的誤區(qū)是「一把大鎖走天下」所有共享訪問都套同一個(gè)Lock結(jié)果是鎖范圍過大、線程全都排隊(duì)多線程退化成單線程還容易死鎖。另一類誤區(qū)是反過來用普通列表當(dāng)隊(duì)列、兩邊一讀一寫偶發(fā)RuntimeError或數(shù)據(jù)丟失。本文不講語法表而是給五個(gè)真實(shí)形狀的場景計(jì)數(shù)器、生產(chǎn)者-消費(fèi)者、優(yōu)雅停止、限流、分階段匯合每個(gè)場景配一個(gè)可運(yùn)行的例子并說明為什么選這種原語而不是別的。示例以 CPython 3.8 及以上為基準(zhǔn)。一、場景共享計(jì)數(shù)器 —— 用Lock問題形狀多個(gè)線程要給同一個(gè)計(jì)數(shù)累加讀-改-寫不能被打斷。# 適用于 Python 3.8import threadingcounter 0counter_lock threading.Lock()def increase(times):global counterfor _ in range(times):with counter_lock: # 臨界區(qū)只包住真正需要保護(hù)的三步counter 1threads [threading.Thread(targetincrease, args(50_000,)) for _ in range(4)]for t in threads:t.start()for t in threads:t.join()print(counter) # 400000為什么用Lock這里需要的是互斥——同一時(shí)刻只允許一個(gè)線程進(jìn)入臨界區(qū)。Lock是最輕的選擇。要點(diǎn)是把臨界區(qū)縮到最小把print、文件讀寫這些不需要保護(hù)的操作挪到鎖外面。什么時(shí)候改RLock如果同一個(gè)線程可能嵌套獲取同一把鎖例如一個(gè)被鎖保護(hù)的函數(shù)內(nèi)部又調(diào)用了另一個(gè)同樣加鎖的函數(shù)Lock會讓它自己把自己鎖死這時(shí)必須用RLock。注意RLock只能由加鎖的那個(gè)線程釋放。二、場景生產(chǎn)者-消費(fèi)者 —— 用queue.Queue問題形狀一批線程產(chǎn)出數(shù)據(jù)另一批線程消費(fèi)數(shù)據(jù)兩邊速度不匹配。這是最經(jīng)典也最容易寫錯(cuò)的場景。手寫listCondition雖然可以但queue.Queue已經(jīng)把鎖、阻塞、喚醒、計(jì)數(shù)全封裝好了。# 適用于 Python 3.8import queueimport threadingimport timeq queue.Queue(maxsize5) # 隊(duì)列上限 5滿了生產(chǎn)者會阻塞def producer(n):for i in range(n):item f數(shù)據(jù)-{i}q.put(item) # 隊(duì)列滿時(shí)自動(dòng)阻塞print(f生產(chǎn) {item})time.sleep(0.02)def consumer(name):while True:item q.get() # 隊(duì)列空時(shí)自動(dòng)阻塞if item is None: # 約定None 表示收工q.task_done()breakprint(f[{name}] 消費(fèi) {item})time.sleep(0.05)q.task_done() # 告訴隊(duì)列這一條處理完了p threading.Thread(targetproducer, args(10,))c threading.Thread(targetconsumer, args(C1,), nameC1)p.start()c.start()p.join()q.put(None) # 發(fā)送結(jié)束信號c.join()q.join() # 等所有 task_doneprint(全部完成)為什么用Queuequeue模塊文檔明確寫著Queue已經(jīng)「實(shí)現(xiàn)了所有必需的加鎖語義」并且可以安全地在多個(gè)生產(chǎn)者和多個(gè)消費(fèi)者之間傳遞。task_done()與join()配對用于等「所有取出的任務(wù)都被處理完」。兩個(gè)關(guān)鍵細(xì)節(jié)一是用maxsize做背壓防止生產(chǎn)太快把內(nèi)存吃光二是結(jié)束信號要選一個(gè)業(yè)務(wù)數(shù)據(jù)中不可能出現(xiàn)的值示例里用的None就是常見約定。三、場景優(yōu)雅停止與啟動(dòng)?xùn)艡?—— 用Event問題形狀一個(gè)線程干活另一個(gè)線程要能讓它停下來或者多個(gè)線程要一起「等口令」再開始。# 適用于 Python 3.8import threadingimport timestop_event threading.Event()start_event threading.Event()def worker(name):print(f{name} 等待啟動(dòng)口令)start_event.wait() # 阻塞直到被別人 set()while not stop_event.is_set():time.sleep(0.05)print(f{name} 工作中……)print(f{name} 已停止)ts [threading.Thread(targetworker, args(fW{i},)) for i in range(3)]for t in ts:t.start()time.sleep(0.1)start_event.set() # 一聲令下三個(gè)線程同時(shí)開始time.sleep(0.2)stop_event.set() # 通知大家收工for t in ts:t.join()為什么用EventEvent管的是一個(gè)布爾標(biāo)志set()會喚醒所有等待的線程wait(timeoutNone)返回True被置位或False超時(shí)。它比Condition更簡單因?yàn)樗恍枰瘸钟墟i。這類「廣播一個(gè)狀態(tài)」的場景用Event最貼切。常用技巧把while not stop.wait(timeout1)寫進(jìn)循環(huán)就同時(shí)獲得了「定時(shí)輪詢」和「立即響應(yīng)停止信號」兩個(gè)效果比while not stop: sleep(1)強(qiáng)得多。四、場景限制并發(fā)數(shù) —— 用Semaphore問題形狀有一批任務(wù)但同一時(shí)刻最多只允許若干個(gè)同時(shí)訪問某個(gè)有限資源比如數(shù)據(jù)庫連接數(shù)、目標(biāo)站點(diǎn)的并發(fā)上限。Lock的計(jì)數(shù)只有 0 和 1Semaphore把計(jì)數(shù)擴(kuò)展到 N。# 適用于 Python 3.8import threadingimport timepool_limit threading.Semaphore(3) # 同時(shí)最多 3 個(gè)線程進(jìn)入def access_db(i):with pool_limit: # 計(jì)數(shù)為 0 時(shí)在這里阻塞print(f任務(wù) {i} 拿到名額)time.sleep(0.2) # 模擬一次數(shù)據(jù)庫訪問threads [threading.Thread(targetaccess_db, args(i,)) for i in range(8)]for t in threads:t.start()for t in threads:t.join()print(全部結(jié)束)為什么用Semaphore它管理的計(jì)數(shù)器表示「release 次數(shù)減 acquire 次數(shù)再加上初值」acquire()在必要時(shí)阻塞保證計(jì)數(shù)不會被減成負(fù)數(shù)。8 個(gè)任務(wù)會分批放行每批最多 3 個(gè)。一個(gè)更嚴(yán)格的選擇是BoundedSemaphore它會在release()次數(shù)超過初值時(shí)拋ValueError。因?yàn)椤羔尫哦嗔恕箮缀跻欢ㄊ谴a bug用有界信號量能把這種 bug 盡早暴露出來——官方文檔也是這么建議的。五、場景分階段匯合 —— 用Barrier問題形狀多個(gè)線程分成若干階段執(zhí)行必須所有線程都完成上一階段才能一起進(jìn)入下一階段。# 適用于 Python 3.8import threadingimport timebarrier threading.Barrier(3)def stage_worker(name):print(f{name} 完成第一階段數(shù)據(jù)準(zhǔn)備)time.sleep(0.05)index barrier.wait() # 等齊 3 個(gè)線程才一起通過print(f{name} 通過柵欄序號 {index}開始第二階段)ts [threading.Thread(targetstage_worker, args(fW{i},)) for i in range(3)]for t in ts:t.start()for t in ts:t.join()為什么用Barrier它專為「N 個(gè)線程必須全部到齊才放行」設(shè)計(jì)。wait()返回 0 到parties-1之間的整數(shù)每個(gè)線程不同可以用來指定一個(gè)線程做收尾工作比如if index 0:打印匯總。如果有一個(gè)線程超時(shí)或abort()柵欄進(jìn)入 broken 狀態(tài)其他等待的線程會收到BrokenBarrierError。六、怎么選一張對照表場景形狀首選原語為什么保護(hù)一段共享讀改Lock最簡單、開銷最小同一線程嵌套加鎖RLock可重入避免自鎖死多生產(chǎn)多消費(fèi)傳數(shù)據(jù)queue.Queue自帶鎖與阻塞語義等待某個(gè)條件成立Condition可等待復(fù)雜謂詞wait_for廣播一個(gè)狀態(tài)/優(yōu)雅停止Event簡單布爾標(biāo)志可喚醒全部限制同時(shí)訪問的線程數(shù)Semaphore/BoundedSemaphore計(jì)數(shù)可大于 1全部到齊才繼續(xù)Barrier分階段同步一個(gè)通用原則能用Queue傳數(shù)據(jù)就不要用共享變量加鎖。隊(duì)列把「同步」這件事收斂到一個(gè)被反復(fù)測試過的實(shí)現(xiàn)里比手寫鎖安全得多。常見坑點(diǎn)1. 用普通列表當(dāng)隊(duì)列兩頭并發(fā)讀寫? 一個(gè)線程lst.append(x)另一個(gè)線程while lst: lst.pop(0)偶發(fā)異?；驍?shù)據(jù)錯(cuò)亂。 ? 改用queue.Queue它內(nèi)部自帶鎖。2. 鎖的范圍包太大? 把print、網(wǎng)絡(luò)請求、文件讀寫全都塞進(jìn)with lock:。 ? 只把真正的讀-改-寫放進(jìn)臨界區(qū)其余挪到鎖外。3. 用Lock卻寫了嵌套加鎖? 持鎖期間又去acquire()同一把Lock線程永久卡住。 ? 需要重入就用RLock更好的做法是理順調(diào)用層次避免嵌套加鎖。4. 忘記task_done()q.join()永遠(yuǎn)不返回? 消費(fèi)者取走數(shù)據(jù)卻沒調(diào)用task_done()主線程q.join()掛死。 ? 每處理完一條就task_done()包括收到結(jié)束信號那一次。5. 用time.sleep輪詢停止信號?while not stop.is_set(): time.sleep(1)最多延遲 1 秒才停。 ?while not stop.wait(timeout1):set()時(shí)立刻返回。6.Semaphore釋放次數(shù)多于獲取次數(shù)? 用Semaphore卻多調(diào)了release()計(jì)數(shù)虛高限流形同虛設(shè)。 ? 需要嚴(yán)格保護(hù)資源上限時(shí)改用BoundedSemaphore讓超額釋放直接報(bào)錯(cuò)。7. 用Event代替Lock做互斥? 兩個(gè)線程用Event互相等待對方set()寫成復(fù)雜易錯(cuò)的交替邏輯。 ?Event是狀態(tài)廣播不是互斥量互斥請用Lock。8. 多把鎖以不同順序獲取造成死鎖? 線程 A 先拿鎖 1 再拿鎖 2線程 B 反過來互相等對方釋放。 ? 全局約定統(tǒng)一的加鎖順序能不嵌套就不嵌套?？偨Y(jié)原語解決的形狀一句話記憶Lock互斥同一時(shí)刻只許一個(gè)進(jìn)RLock可重入互斥同一線程可以反復(fù)進(jìn)Condition等條件等謂詞成立再走Event廣播狀態(tài)一聲令下全體響應(yīng)Semaphore限流同時(shí)最多 N 個(gè)Barrier匯合到齊了才放行queue.Queue傳遞數(shù)據(jù)自帶鎖的安全管道同步機(jī)制的選型本質(zhì)是先看清「問題形狀」再挑對應(yīng)的原語。互斥、傳值、廣播、限流、匯合是五種完全不同的形狀用錯(cuò)了形狀即使語法正確也會寫出又慢又容易死鎖的代碼。實(shí)在拿不準(zhǔn)時(shí)優(yōu)先退回到queue.Queue加線程池的組合——它覆蓋了大多數(shù)真實(shí)需求且比手寫鎖更難出錯(cuò)。 Python線程之同步機(jī)制實(shí)際應(yīng)用場景舉例說明
前言線程共享內(nèi)存所以只要兩個(gè)線程碰同一個(gè)可變對象就必須想清楚「怎么同步」。但同步不等于「到處加鎖」——threading模塊提供了五六種原語每種解決的是不同形狀的協(xié)調(diào)問題有的保護(hù)一段臨界區(qū)有的負(fù)責(zé)線程之間傳消息有的只是發(fā)個(gè)信號還有的是限制同時(shí)干活的人數(shù)。新手最常見的誤區(qū)是「一把大鎖走天下」所有共享訪問都套同一個(gè)Lock結(jié)果是鎖范圍過大、線程全都排隊(duì)多線程退化成單線程還容易死鎖。另一類誤區(qū)是反過來用普通列表當(dāng)隊(duì)列、兩邊一讀一寫偶發(fā)RuntimeError或數(shù)據(jù)丟失。本文不講語法表而是給五個(gè)真實(shí)形狀的場景計(jì)數(shù)器、生產(chǎn)者-消費(fèi)者、優(yōu)雅停止、限流、分階段匯合每個(gè)場景配一個(gè)可運(yùn)行的例子并說明為什么選這種原語而不是別的。示例以 CPython 3.8 及以上為基準(zhǔn)。一、場景共享計(jì)數(shù)器 —— 用Lock問題形狀多個(gè)線程要給同一個(gè)計(jì)數(shù)累加讀-改-寫不能被打斷。# 適用于 Python 3.8import threadingcounter 0counter_lock threading.Lock()def increase(times):global counterfor _ in range(times):with counter_lock: # 臨界區(qū)只包住真正需要保護(hù)的三步counter 1threads [threading.Thread(targetincrease, args(50_000,)) for _ in range(4)]for t in threads:t.start()for t in threads:t.join()print(counter) # 400000為什么用Lock這里需要的是互斥——同一時(shí)刻只允許一個(gè)線程進(jìn)入臨界區(qū)。Lock是最輕的選擇。要點(diǎn)是把臨界區(qū)縮到最小把print、文件讀寫這些不需要保護(hù)的操作挪到鎖外面。什么時(shí)候改RLock如果同一個(gè)線程可能嵌套獲取同一把鎖例如一個(gè)被鎖保護(hù)的函數(shù)內(nèi)部又調(diào)用了另一個(gè)同樣加鎖的函數(shù)Lock會讓它自己把自己鎖死這時(shí)必須用RLock。注意RLock只能由加鎖的那個(gè)線程釋放。二、場景生產(chǎn)者-消費(fèi)者 —— 用queue.Queue問題形狀一批線程產(chǎn)出數(shù)據(jù)另一批線程消費(fèi)數(shù)據(jù)兩邊速度不匹配。這是最經(jīng)典也最容易寫錯(cuò)的場景。手寫listCondition雖然可以但queue.Queue已經(jīng)把鎖、阻塞、喚醒、計(jì)數(shù)全封裝好了。# 適用于 Python 3.8import queueimport threadingimport timeq queue.Queue(maxsize5) # 隊(duì)列上限 5滿了生產(chǎn)者會阻塞def producer(n):for i in range(n):item f數(shù)據(jù)-{i}q.put(item) # 隊(duì)列滿時(shí)自動(dòng)阻塞print(f生產(chǎn) {item})time.sleep(0.02)def consumer(name):while True:item q.get() # 隊(duì)列空時(shí)自動(dòng)阻塞if item is None: # 約定None 表示收工q.task_done()breakprint(f[{name}] 消費(fèi) {item})time.sleep(0.05)q.task_done() # 告訴隊(duì)列這一條處理完了p threading.Thread(targetproducer, args(10,))c threading.Thread(targetconsumer, args(C1,), nameC1)p.start()c.start()p.join()q.put(None) # 發(fā)送結(jié)束信號c.join()q.join() # 等所有 task_doneprint(全部完成)為什么用Queuequeue模塊文檔明確寫著Queue已經(jīng)「實(shí)現(xiàn)了所有必需的加鎖語義」并且可以安全地在多個(gè)生產(chǎn)者和多個(gè)消費(fèi)者之間傳遞。task_done()與join()配對用于等「所有取出的任務(wù)都被處理完」。兩個(gè)關(guān)鍵細(xì)節(jié)一是用maxsize做背壓防止生產(chǎn)太快把內(nèi)存吃光二是結(jié)束信號要選一個(gè)業(yè)務(wù)數(shù)據(jù)中不可能出現(xiàn)的值示例里用的None就是常見約定。三、場景優(yōu)雅停止與啟動(dòng)?xùn)艡?—— 用Event問題形狀一個(gè)線程干活另一個(gè)線程要能讓它停下來或者多個(gè)線程要一起「等口令」再開始。# 適用于 Python 3.8import threadingimport timestop_event threading.Event()start_event threading.Event()def worker(name):print(f{name} 等待啟動(dòng)口令)start_event.wait() # 阻塞直到被別人 set()while not stop_event.is_set():time.sleep(0.05)print(f{name} 工作中……)print(f{name} 已停止)ts [threading.Thread(targetworker, args(fW{i},)) for i in range(3)]for t in ts:t.start()time.sleep(0.1)start_event.set() # 一聲令下三個(gè)線程同時(shí)開始time.sleep(0.2)stop_event.set() # 通知大家收工for t in ts:t.join()為什么用EventEvent管的是一個(gè)布爾標(biāo)志set()會喚醒所有等待的線程wait(timeoutNone)返回True被置位或False超時(shí)。它比Condition更簡單因?yàn)樗恍枰瘸钟墟i。這類「廣播一個(gè)狀態(tài)」的場景用Event最貼切。常用技巧把while not stop.wait(timeout1)寫進(jìn)循環(huán)就同時(shí)獲得了「定時(shí)輪詢」和「立即響應(yīng)停止信號」兩個(gè)效果比while not stop: sleep(1)強(qiáng)得多。四、場景限制并發(fā)數(shù) —— 用Semaphore問題形狀有一批任務(wù)但同一時(shí)刻最多只允許若干個(gè)同時(shí)訪問某個(gè)有限資源比如數(shù)據(jù)庫連接數(shù)、目標(biāo)站點(diǎn)的并發(fā)上限。Lock的計(jì)數(shù)只有 0 和 1Semaphore把計(jì)數(shù)擴(kuò)展到 N。# 適用于 Python 3.8import threadingimport timepool_limit threading.Semaphore(3) # 同時(shí)最多 3 個(gè)線程進(jìn)入def access_db(i):with pool_limit: # 計(jì)數(shù)為 0 時(shí)在這里阻塞print(f任務(wù) {i} 拿到名額)time.sleep(0.2) # 模擬一次數(shù)據(jù)庫訪問threads [threading.Thread(targetaccess_db, args(i,)) for i in range(8)]for t in threads:t.start()for t in threads:t.join()print(全部結(jié)束)為什么用Semaphore它管理的計(jì)數(shù)器表示「release 次數(shù)減 acquire 次數(shù)再加上初值」acquire()在必要時(shí)阻塞保證計(jì)數(shù)不會被減成負(fù)數(shù)。8 個(gè)任務(wù)會分批放行每批最多 3 個(gè)。一個(gè)更嚴(yán)格的選擇是BoundedSemaphore它會在release()次數(shù)超過初值時(shí)拋ValueError。因?yàn)椤羔尫哦嗔恕箮缀跻欢ㄊ谴a bug用有界信號量能把這種 bug 盡早暴露出來——官方文檔也是這么建議的。五、場景分階段匯合 —— 用Barrier問題形狀多個(gè)線程分成若干階段執(zhí)行必須所有線程都完成上一階段才能一起進(jìn)入下一階段。# 適用于 Python 3.8import threadingimport timebarrier threading.Barrier(3)def stage_worker(name):print(f{name} 完成第一階段數(shù)據(jù)準(zhǔn)備)time.sleep(0.05)index barrier.wait() # 等齊 3 個(gè)線程才一起通過print(f{name} 通過柵欄序號 {index}開始第二階段)ts [threading.Thread(targetstage_worker, args(fW{i},)) for i in range(3)]for t in ts:t.start()for t in ts:t.join()為什么用Barrier它專為「N 個(gè)線程必須全部到齊才放行」設(shè)計(jì)。wait()返回 0 到parties-1之間的整數(shù)每個(gè)線程不同可以用來指定一個(gè)線程做收尾工作比如if index 0:打印匯總。如果有一個(gè)線程超時(shí)或abort()柵欄進(jìn)入 broken 狀態(tài)其他等待的線程會收到BrokenBarrierError。六、怎么選一張對照表場景形狀首選原語為什么保護(hù)一段共享讀改Lock最簡單、開銷最小同一線程嵌套加鎖RLock可重入避免自鎖死多生產(chǎn)多消費(fèi)傳數(shù)據(jù)queue.Queue自帶鎖與阻塞語義等待某個(gè)條件成立Condition可等待復(fù)雜謂詞wait_for廣播一個(gè)狀態(tài)/優(yōu)雅停止Event簡單布爾標(biāo)志可喚醒全部限制同時(shí)訪問的線程數(shù)Semaphore/BoundedSemaphore計(jì)數(shù)可大于 1全部到齊才繼續(xù)Barrier分階段同步一個(gè)通用原則能用Queue傳數(shù)據(jù)就不要用共享變量加鎖。隊(duì)列把「同步」這件事收斂到一個(gè)被反復(fù)測試過的實(shí)現(xiàn)里比手寫鎖安全得多。常見坑點(diǎn)1. 用普通列表當(dāng)隊(duì)列兩頭并發(fā)讀寫? 一個(gè)線程lst.append(x)另一個(gè)線程while lst: lst.pop(0)偶發(fā)異?；驍?shù)據(jù)錯(cuò)亂。 ? 改用queue.Queue它內(nèi)部自帶鎖。2. 鎖的范圍包太大? 把print、網(wǎng)絡(luò)請求、文件讀寫全都塞進(jìn)with lock:。 ? 只把真正的讀-改-寫放進(jìn)臨界區(qū)其余挪到鎖外。3. 用Lock卻寫了嵌套加鎖? 持鎖期間又去acquire()同一把Lock線程永久卡住。 ? 需要重入就用RLock更好的做法是理順調(diào)用層次避免嵌套加鎖。4. 忘記task_done()q.join()永遠(yuǎn)不返回? 消費(fèi)者取走數(shù)據(jù)卻沒調(diào)用task_done()主線程q.join()掛死。 ? 每處理完一條就task_done()包括收到結(jié)束信號那一次。5. 用time.sleep輪詢停止信號?while not stop.is_set(): time.sleep(1)最多延遲 1 秒才停。 ?while not stop.wait(timeout1):set()時(shí)立刻返回。6.Semaphore釋放次數(shù)多于獲取次數(shù)? 用Semaphore卻多調(diào)了release()計(jì)數(shù)虛高限流形同虛設(shè)。 ? 需要嚴(yán)格保護(hù)資源上限時(shí)改用BoundedSemaphore讓超額釋放直接報(bào)錯(cuò)。7. 用Event代替Lock做互斥? 兩個(gè)線程用Event互相等待對方set()寫成復(fù)雜易錯(cuò)的交替邏輯。 ?Event是狀態(tài)廣播不是互斥量互斥請用Lock。8. 多把鎖以不同順序獲取造成死鎖? 線程 A 先拿鎖 1 再拿鎖 2線程 B 反過來互相等對方釋放。 ? 全局約定統(tǒng)一的加鎖順序能不嵌套就不嵌套?？偨Y(jié)原語解決的形狀一句話記憶Lock互斥同一時(shí)刻只許一個(gè)進(jìn)RLock可重入互斥同一線程可以反復(fù)進(jìn)Condition等條件等謂詞成立再走Event廣播狀態(tài)一聲令下全體響應(yīng)Semaphore限流同時(shí)最多 N 個(gè)Barrier匯合到齊了才放行queue.Queue傳遞數(shù)據(jù)自帶鎖的安全管道同步機(jī)制的選型本質(zhì)是先看清「問題形狀」再挑對應(yīng)的原語。互斥、傳值、廣播、限流、匯合是五種完全不同的形狀用錯(cuò)了形狀即使語法正確也會寫出又慢又容易死鎖的代碼。實(shí)在拿不準(zhǔn)時(shí)優(yōu)先退回到queue.Queue加線程池的組合——它覆蓋了大多數(shù)真實(shí)需求且比手寫鎖更難出錯(cuò)。