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