：從線程創(chuàng)建到GIL鎖機(jī)制詳解)
在 Python 進(jìn)階路上多線程幾乎是繞不開的一道坎。很多人學(xué)了函數(shù)、類、文件操作之后一旦遇到“下載一批圖片”“同時處理多個請求”“爬蟲抓取多個頁面”這類需求就會發(fā)現(xiàn)單線程程序慢得讓人著急。第 37 天我們來把 Python 的threading模塊徹底講透從一個線程怎么創(chuàng)建到鎖怎么加再到真實場景里到底該不該用多線程全部用代碼過一遍。這篇教程會圍繞幾個核心問題展開Python 里線程和進(jìn)程到底什么關(guān)系、threading模塊怎么用、join和daemon是什么、多線程訪問共享數(shù)據(jù)為什么會出問題、Lock該怎么加以及 GIL 對 Python 多線程到底有什么影響。每個知識點(diǎn)都會給出可以直接運(yùn)行的代碼示例建議讀者順手打開自己的 Python 環(huán)境跟著敲一遍。文末還會附上多線程面試常見問題和工程實踐建議方便對照排查自己的知識盲區(qū)。1. 核心知識點(diǎn)速覽Python 多線程與 threading 模塊主題說明項目定位Python 標(biāo)準(zhǔn)庫多線程編程教程核心模塊threadingPython 內(nèi)置無需 pip 安裝適用 Python 版本Python 3.x3.6 均可推薦 3.8前置知識Python 基礎(chǔ)語法、函數(shù)、類、列表字典操作學(xué)習(xí)目標(biāo)掌握線程創(chuàng)建、啟動、等待、守護(hù)線程、鎖機(jī)制典型應(yīng)用I/O 密集型任務(wù)、網(wǎng)絡(luò)請求、文件讀寫、爬蟲不適合場景CPU 密集型計算受 GIL 限制建議用 multiprocessing運(yùn)行平臺Windows / Linux / macOS 均可運(yùn)行這里先給一個結(jié)論P(yáng)ython 的threading模塊解決的是“并發(fā)”問題不是“并行”問題。它適合讓程序在等待網(wǎng)絡(luò)響應(yīng)、磁盤讀寫時去執(zhí)行別的任務(wù)從而大幅縮短總耗時。2. 使用場景與學(xué)習(xí)邊界什么時候?qū)W多線程什么時候別用多線程不是銀彈學(xué)之前先搞清楚它能解決什么問題也要知道它不適合解決什么問題。2.1 適合用多線程的場景I/O 密集型任務(wù)是 Python 多線程的主場。所謂 I/O 密集型指程序大部分時間都在等待外部設(shè)備返回數(shù)據(jù)CPU 本身反而比較空閑。典型場景包括網(wǎng)絡(luò)爬蟲批量請求網(wǎng)頁每個請求要等服務(wù)器返回。批量下載圖片、文件、視頻。從多個 API 接口拉取數(shù)據(jù)。讀寫大量小文件磁盤 I/O 頻繁但數(shù)據(jù)量不大。數(shù)據(jù)庫查詢量大每次查詢都有網(wǎng)絡(luò)開銷。同時運(yùn)行多個定時任務(wù)或監(jiān)控任務(wù)。在這些場景中一個線程發(fā)起請求后阻塞等待另一個線程立刻被調(diào)度執(zhí)行程序總耗時可縮短到原來的幾分之一甚至幾十分之一。2.2 不適合用多線程的場景CPU 密集型任務(wù)不適合用 Python 多線程。例如大量數(shù)學(xué)計算、圖像像素循環(huán)處理、復(fù)雜算法遞歸、大數(shù)據(jù)排序等。這類任務(wù)需要 CPU 持續(xù)高強(qiáng)度工作而 Python 的 GIL 全局解釋器鎖會讓同一時刻只有一個線程在執(zhí)行 Python 字節(jié)碼多線程反而可能因為線程切換增加額外開銷速度不升反降。CPU 密集場景的正確選擇是multiprocessing進(jìn)程池讓多個進(jìn)程分別占用不同的 CPU 核心。這是第 38 天之后會展開的內(nèi)容本文先不深入。2.3 學(xué)習(xí)路徑建議建議按以下順序?qū)W完本篇文章弄清進(jìn)程與線程的區(qū)別。學(xué)會用threading.Thread創(chuàng)建線程。掌握start()、join()、daemon的使用。理解共享數(shù)據(jù)的線程安全問題。學(xué)會用Lock解決競態(tài)條件。用兩個綜合例程驗證所學(xué)。了解 GIL 對多線程的底層影響。最后做一組面試題自測。3. 環(huán)境準(zhǔn)備Python 多線程編程的前置條件3.1 安裝 Pythonthreading是 Python 標(biāo)準(zhǔn)庫不需要額外安裝。你只需要一個能運(yùn)行 Python 3 的環(huán)境即可。Windows 用戶建議直接到 Python 官網(wǎng)下載安裝包安裝時務(wù)必勾選“Add Python to PATH”。如果安裝后命令行輸入python沒有反應(yīng)說明環(huán)境變量沒有配置好需要手動把 Python 安裝路徑加入 PATH。macOS 用戶建議使用 Homebrew 安裝brew install pythonLinux 用戶多數(shù)發(fā)行版自帶 Python 3可直接使用。沒有的話執(zhí)行sudo apt update sudo apt install python3驗證環(huán)境python --version能輸出 Python 3.x 版本號就說明環(huán)境沒問題。為了代碼編輯體驗更好建議安裝 VS Code 并配置 Python 擴(kuò)展或者直接使用 PyCharm 社區(qū)版。VS Code 里新建.py文件后按CtrlF5即可直接運(yùn)行。3.2 驗證 threading 模塊可用在 Python 交互式環(huán)境或新建腳本中執(zhí)行import threading print(threading.__version__) # Python 3.10 不再暴露版本號會報 AttributeError print(threading.current_thread().name)threading模塊在 Python 3.x 中已經(jīng)非常穩(wěn)定。輸出當(dāng)前線程名稱為MainThread即說明模塊正常工作。3.3 基礎(chǔ)概念準(zhǔn)備寫多線程代碼前先把這幾個核心概念搞清楚。進(jìn)程操作系統(tǒng)分配資源的基本單位。一個程序啟動后就是一個進(jìn)程擁有獨(dú)立的內(nèi)存空間。線程進(jìn)程內(nèi)部的任務(wù)執(zhí)行單元。一個進(jìn)程可以包含多個線程線程之間共享進(jìn)程的內(nèi)存空間。并發(fā)多個任務(wù)交替執(zhí)行看起來像是同時進(jìn)行實際在單核 CPU 上是通過時間片切換完成的。并行多個任務(wù)真的在同一時刻同時執(zhí)行需要多核 CPU 支持。Python 中的 GIL 決定 CPython 解釋器同一時間只允許一個線程執(zhí)行 Python 字節(jié)碼。所以 Python 的多線程更像是一種“并發(fā)”工具用于在等待 I/O 時切換到其他任務(wù)而不是把計算任務(wù)分給多個 CPU 核心并行處理。4. threading 模塊核心用法與代碼實戰(zhàn)4.1 最簡單的線程創(chuàng)建方式直接實例化 Threadthreading.Thread是threading模塊最常用的類。創(chuàng)建線程有兩種方式直接傳入函數(shù)或傳入可調(diào)用對象。第一種方式把普通函數(shù)作為target傳入import threading import time def work(): print(f子線程 {threading.current_thread().name} 開始執(zhí)行) time.sleep(2) print(f子線程 {threading.current_thread().name} 執(zhí)行完畢) print(f主線程 {threading.current_thread().name} 開始) # 創(chuàng)建線程target 指向要執(zhí)行的函數(shù) t threading.Thread(targetwork, nameWorkerThread) # 啟動線程 t.start() print(主線程繼續(xù)往下執(zhí)行)運(yùn)行這段代碼控制臺輸出順序可能類似主線程 MainThread 開始 主線程繼續(xù)往下執(zhí)行 子線程 WorkerThread 開始執(zhí)行 等待 2 秒 子線程 WorkerThread 執(zhí)行完畢注意觀察輸出順序調(diào)用start()之后主線程并不會阻塞等待子線程結(jié)束而是繼續(xù)往下執(zhí)行。子線程什么時候真正被調(diào)度執(zhí)行由操作系統(tǒng)的線程調(diào)度器決定。你可能會發(fā)現(xiàn)WorkerThread 開始執(zhí)行這一行在主線程繼續(xù)往下執(zhí)行之前或之后出現(xiàn)這完全正常。如果你希望主線程等子線程執(zhí)行完畢后再繼續(xù)就需要用到j(luò)oin()import threading import time def work(): print(f子線程 {threading.current_thread().name} 開始執(zhí)行) time.sleep(2) print(f子線程 {threading.current_thread().name} 執(zhí)行完畢) t threading.Thread(targetwork, nameWorkerThread) t.start() # 等待子線程執(zhí)行完畢 t.join() print(主線程在子線程結(jié)束后繼續(xù)執(zhí)行)加上join()后輸出順序變成固定的子線程先完整執(zhí)行完主線程再打最后一行。4.2 自定義線程類繼承 Thread 并重寫 run除了直接傳入函數(shù)還可以通過繼承Thread類并重寫run()方法創(chuàng)建線程。當(dāng)線程調(diào)用start()時內(nèi)部會自動調(diào)用run()方法。這種方式適合邏輯比較復(fù)雜的場景可以把線程相關(guān)的屬性封裝在類中import threading import time class DownloadThread(threading.Thread): def __init__(self, file_name, url): super().__init__() self.file_name file_name self.url url def run(self): print(f[{self.name}] 開始下載 {self.file_name}地址: {self.url}) time.sleep(3) print(f[{self.name}] {self.file_name} 下載完成) if __name__ __main__: tasks [ (圖片1.jpg, https://example.com/img1.jpg), (圖片2.jpg, https://example.com/img2.jpg), (圖片3.jpg, https://example.com/img3.jpg), ] threads [] for name, url in tasks: t DownloadThread(name, url) threads.append(t) t.start() # 等待所有線程執(zhí)行完畢 for t in threads: t.join() print(所有下載任務(wù)已完成)這里self.name是Thread類的內(nèi)置屬性線程未手動命名時系統(tǒng)會自動分配Thread-1、Thread-2這樣的名稱。4.3 線程生命周期與 daemon 守護(hù)線程Python 程序運(yùn)行邏輯中有個重要規(guī)則主線程結(jié)束時如果仍有非守護(hù)線程存活程序會等待所有非守護(hù)線程結(jié)束才退出如果只剩下守護(hù)線程程序會直接結(jié)束。守護(hù)線程通過daemonTrue設(shè)置。典型應(yīng)用是后臺日志收集、心跳檢測、垃圾回收輔助線程等。先看非守護(hù)線程的行為import threading import time def work(): print(子線程開始) time.sleep(3) print(子線程結(jié)束) t threading.Thread(targetwork) t.start() print(主線程即將結(jié)束)輸出順序是子線程開始 主線程即將結(jié)束 程序停留約 3 秒 子線程結(jié)束因為t不是守護(hù)線程主線程執(zhí)行完后必須等t執(zhí)行完程序才會結(jié)束。再看守護(hù)線程import threading import time def work(): print(守護(hù)線程開始) time.sleep(3) print(守護(hù)線程結(jié)束) t threading.Thread(targetwork, daemonTrue) t.start() print(主線程即將結(jié)束)輸出順序可能變成守護(hù)線程開始 主線程即將結(jié)束 程序立即退出守護(hù)線程的結(jié)束語句很可能根本來不及打印因為主線程結(jié)束后程序直接退出不會等守護(hù)線程。實際開發(fā)中守護(hù)線程常用于線程池中的后臺清理任務(wù)?！爱?dāng)主線程結(jié)束守護(hù)線程自動銷毀”這個特性非常方便但要注意守護(hù)線程中的資源清理邏輯不能依賴主線程的返回值。4.4 多線程訪問共享數(shù)據(jù)線程安全問題初現(xiàn)多線程最大的坑不是創(chuàng)建線程而是多個線程同時修改共享數(shù)據(jù)導(dǎo)致結(jié)果錯誤。來看一個經(jīng)典例子import threading count 0 def add(): global count for _ in range(1000000): count 1 def sub(): global count for _ in range(1000000): count - 1 t1 threading.Thread(targetadd) t2 threading.Thread(targetsub) t1.start() t2.start() t1.join() t2.join() print(f最終 count {count})如果兩個線程嚴(yán)格串行執(zhí)行count最終應(yīng)該還是0因為加一百萬次、減一百萬次相互抵消。但實際多次運(yùn)行這段代碼結(jié)果幾乎不可能是 0而是一個正數(shù)或負(fù)數(shù)。這就是典型的競態(tài)條件。原因在于count 1這一行代碼在 Python 底層不是一步完成的它至少包含三條 CPU 指令讀取count當(dāng)前值。執(zhí)行加法運(yùn)算。把新值寫回count。線程 A 執(zhí)行完第 2 步還沒寫回時線程 B 也可能讀取到舊的count值并開始計算。兩個線程同時基于同一個舊值做運(yùn)算最后寫回的新值就相互覆蓋。很多次的更新操作白白丟失最終結(jié)果就偏離了預(yù)期。4.5 Lock 鎖解決競態(tài)條件threading.Lock提供互斥鎖機(jī)制保證同一時刻只有一個線程能進(jìn)入臨界區(qū)。加鎖后的代碼段成為“原子操作”其他線程必須等待鎖被釋放才能進(jìn)入。改進(jìn)后的代碼import threading count 0 lock threading.Lock() def add(): global count for _ in range(1000000): with lock: count 1 def sub(): global count for _ in range(1000000): with lock: count - 1 t1 threading.Thread(targetadd) t2 threading.Thread(targetsub) t1.start() t2.start() t1.join() t2.join() print(f最終 count {count})輸出結(jié)果每次都是穩(wěn)定的0。Lock有兩種使用方式方式一手動調(diào)用acquire()和release()lock.acquire() try: count 1 finally: lock.release()方式二使用with語句推薦的寫法with lock: count 1第二種寫法可以保證即使代碼拋出異常鎖也會被正確釋放。日常開發(fā)中建議只用with方式。關(guān)于鎖還有一些細(xì)節(jié)值得留意。threading.Lock()是可重入的普通鎖嗎答案是普通Lock不可重入。如果同一個線程在未釋放鎖的情況下再次調(diào)用acquire()程序會死鎖。需要可重入鎖的話應(yīng)使用threading.RLock()。另外加鎖范圍越大并發(fā)效率越低。合理做法是只鎖定“真正需要保護(hù)的那幾行代碼”不要在整個循環(huán)外包一層鎖否則兩個線程依然會變成串行執(zhí)行失去多線程的意義。4.6 線程池concurrent.futures 與 ThreadPoolExecutor雖然threading模塊是本文核心但必須提前介紹線程池。實際項目里直接手動創(chuàng)建幾十個裸線程并不是好方案線程的創(chuàng)建和銷毀都有開銷大量線程同時運(yùn)行反而會導(dǎo)致頻繁上下文切換CPU 使用率飆升但任務(wù)完成速度反而下降。Python 推薦用concurrent.futures.ThreadPoolExecutor管理線程from concurrent.futures import ThreadPoolExecutor import time def fetch_data(url): print(f開始請求: {url}) time.sleep(2) return f{url} 的數(shù)據(jù) urls [ https://api.example.com/user/1, https://api.example.com/user/2, https://api.example.com/user/3, https://api.example.com/user/4, https://api.example.com/user/5, ] # 創(chuàng)建最多 3 個線程的線程池 with ThreadPoolExecutor(max_workers3) as executor: futures [executor.submit(fetch_data, url) for url in urls] for future in futures: print(future.result())這段代碼中線程池最多同時運(yùn)行 3 個線程5 個任務(wù)按隊列依次執(zhí)行。with塊結(jié)束時線程池會等待所有任務(wù)完成并自動清理資源。比起手動管理Thread對象的start()和join()簡潔非常多。4.7 線程間通信 Queue多線程之間交換數(shù)據(jù)時手工加鎖管理共享列表很容易出錯。Python 的queue.Queue是線程安全的隊列內(nèi)部自己實現(xiàn)了鎖機(jī)制專門用于多線程生產(chǎn)者-消費(fèi)者模型。import threading import queue import time def producer(q): for i in range(5): item f產(chǎn)品-{i} q.put(item) print(f生產(chǎn)了: {item}) time.sleep(1) def consumer(q): while True: item q.get() if item is None: break print(f消費(fèi)了: {item}) q.task_done() q queue.Queue(maxsize3) threads [ threading.Thread(targetproducer, args(q,)), threading.Thread(targetconsumer, args(q,)), ] for t in threads: t.start() q.join() print(所有任務(wù)完成)生產(chǎn)者線程向隊列添加數(shù)據(jù)消費(fèi)者線程從隊列取出數(shù)據(jù)。Queue內(nèi)部已經(jīng)加鎖多線程同時put或get不會出現(xiàn)數(shù)據(jù)錯亂比手動維護(hù)list加Lock可靠得多。5. 綜合代碼實戰(zhàn)批量下載任務(wù)中的多線程應(yīng)用為了把前面所有知識點(diǎn)串起來這里設(shè)計一個貼近真實業(yè)務(wù)的下載任務(wù)案例。假設(shè)需要在 10 秒內(nèi)從模擬服務(wù)器下載 6 個文件單線程需要約 18 秒用多線程可以把耗時壓縮到約 6 秒。import threading import time import random class MockDownloader: 模擬從網(wǎng)絡(luò)下載文件的類 def __init__(self): self.lock threading.Lock() self.completed 0 def download(self, file_id): # 模擬不同文件大小導(dǎo)致的不同下載時長 cost random.uniform(2, 4) time.sleep(cost) with self.lock: self.completed 1 print(f文件 {file_id} 下載完成耗時 {cost:.2f}s總進(jìn)度 {self.completed}/6) def run(self): threads [] for i in range(1, 7): t threading.Thread(targetself.download, args(i,), namefDownload-{i}) threads.append(t) t.start() for t in threads: t.join() print(f全部下載完成共 {self.completed} 個文件) if __name__ __main__: start time.time() MockDownloader().run() print(f總耗時: {time.time() - start:.2f} 秒)這個案例體現(xiàn)了三個重點(diǎn)多個線程同時發(fā)起“下載”、用threading.current_thread().name或自定義線程名區(qū)分日志、用Lock保護(hù)共享進(jìn)度值避免打印統(tǒng)計錯亂。如果這里不用鎖self.completed 1在極端情況下也會出現(xiàn)計數(shù)丟失的問題。雖然 CPython 對整數(shù)自增有一定優(yōu)化但在高并發(fā)場景下依然不可靠工程上默認(rèn)加鎖是最穩(wěn)妥的。6. 多線程性能觀察與 GIL 影響分析6.1 用實驗證明 I/O 密集型任務(wù)的多線程加速寫一個模擬 I/O 等待的程序?qū)Ρ葐尉€程和多線程耗時import threading import time def io_task(): time.sleep(2) # 單線程執(zhí)行 5 次 start time.time() for _ in range(5): io_task() print(f單線程耗時: {time.time() - start:.2f}s) # 多線程執(zhí)行 5 次 start time.time() threads [] for _ in range(5): t threading.Thread(targetio_task) threads.append(t) t.start() for t in threads: t.join() print(f多線程耗時: {time.time() - start:.2f}s)運(yùn)行結(jié)果通常非常接近單線程耗時: 10.01s 多線程耗時: 2.01s5 個線程同時睡眠等待總耗時不疊加說明多線程成功利用了 I/O 等待時間。6.2 CPU 密集型任務(wù)不要用多線程換一個 CPU 密集型實驗import threading import time def cpu_task(): total 0 for i in range(50000000): total i return total start time.time() for _ in range(4): cpu_task() print(f單線程耗時: {time.time() - start:.2f}s) start time.time() threads [] for _ in range(4): t threading.Thread(targetcpu_task) threads.append(t) t.start() for t in threads: t.join() print(f多線程耗時: {time.time() - start:.2f}s)運(yùn)行后會發(fā)現(xiàn)多線程版本很可能比單線程還慢或者只是持平。原因就是 GIL。CPython 解釋器中任何時刻只允許一個線程執(zhí)行 Python 字節(jié)碼CPU 計算任務(wù)無法利用多核優(yōu)勢。線程越多切換開銷越大性能反而下降。CPU 密集型任務(wù)應(yīng)改用多進(jìn)程。這是官方推薦、社區(qū)公認(rèn)的做法。具體代碼會在后續(xù)文章介紹multiprocessing時詳細(xì)展開本文先記住結(jié)論I/O 密集型用threadingCPU 密集型用multiprocessing。6.3 如何觀察線程運(yùn)行狀態(tài)線程不是越多越好。觀察線程數(shù)量最簡單的方式import threading import time def work(): time.sleep(3) threads [] for i in range(10): t threading.Thread(targetwork) threads.append(t) t.start() print(f當(dāng)前活動線程數(shù): {threading.active_count()}) print(f當(dāng)前線程列表: {threading.enumerate()})threading.active_count()返回當(dāng)前存活線程數(shù)量threading.enumerate()列出所有線程對象。調(diào)試時可以通過打印線程名區(qū)分不同線程的輸出import threading def func(): print(f{threading.current_thread().name} 正在執(zhí)行) t threading.Thread(targetfunc, nameMyWorker) t.start()線程名最好在創(chuàng)建時就指定這樣日志可讀性會好很多。7. 線程安全與常見陷阱排查7.1 直接修改共享 list 或 dict多個線程同時append到同一個list、或同時update同一個dict看起來安全但較大數(shù)據(jù)量下仍可能出現(xiàn)狀態(tài)不一致。不要賭 CPython 的原子性使用鎖保護(hù)共享容器是更嚴(yán)謹(jǐn)?shù)淖龇ā?.2 忘記 join 導(dǎo)致數(shù)據(jù)還沒處理完程序就退出核心工作線程是普通線程主線程忘記join()時主線程雖然不會立即退出但某些邏輯如后續(xù)統(tǒng)計匯總會提前執(zhí)行。建議每個start()都有對應(yīng)的join()不要省略。7.3 死鎖鎖順序不一致多個鎖存在時不同線程按不同順序獲取鎖可能導(dǎo)致互相等待。例如線程 A 持有鎖 1 等待鎖 2線程 B 持有鎖 2 等待鎖 1程序永久卡住。避免死鎖的方法是讓所有線程按相同順序獲取鎖或者盡量用單把鎖控制臨界區(qū)。實際業(yè)務(wù)中要避免鎖嵌套鎖的范圍越小越好。7.4 無限制創(chuàng)建線程導(dǎo)致資源耗盡創(chuàng)建大量線程而不復(fù)用每個線程都占用內(nèi)存和系統(tǒng)資源最終導(dǎo)致 OOM 或線程創(chuàng)建失敗。生產(chǎn)環(huán)境統(tǒng)一用ThreadPoolExecutor(max_workersN)限流不建議直接循環(huán)Thread。7.5daemon線程中的資源清理如果守護(hù)線程持有文件句柄或數(shù)據(jù)庫連接主線程退出時資源可能無法正常釋放。因此守護(hù)線程不要承擔(dān)需要完整清理的任務(wù)寧愿用join()等待或獨(dú)立的進(jìn)程服務(wù)。8. 實戰(zhàn)ThreadPoolExecutor 批量 API 數(shù)據(jù)拉取手動管理十幾個Thread已經(jīng)讓人頭疼真實項目中更推薦直接用線程池。這里給出適合做數(shù)據(jù)采集任務(wù)的通用模板from concurrent.futures import ThreadPoolExecutor, as_completed import time import random def fetch_one(user_id): # 模擬調(diào)用外部 API cost random.uniform(0.5, 2) time.sleep(cost) return {user_id: user_id, status: ok, delay: round(cost, 2)} user_ids list(range(1, 11)) # max_workers 一般設(shè)置為 I/O 型的 5~20 之間需根據(jù)目標(biāo)和API速率調(diào)整 with ThreadPoolExecutor(max_workers5) as executor: future_map {executor.submit(fetch_one, uid): uid for uid in user_ids} for future in as_completed(future_map): uid future_map[future] try: result future.result() print(f用戶 {uid}: {result}) except Exception as exc: print(f用戶 {uid} 請求失敗: {exc})重點(diǎn)看兩點(diǎn)as_completed按任務(wù)完成順序返回結(jié)果不必等待前面的慢任務(wù)可以直接把完成的數(shù)據(jù)寫入文件或數(shù)據(jù)庫。future.result()會拋出任務(wù)內(nèi)部的異常調(diào)用方必須捕獲否則一個任務(wù)出錯可能導(dǎo)致整個主線程中斷。真實采集數(shù)據(jù)時建議配合重試機(jī)制和日志輸出。比如單個請求失敗時先記錄下來全部請求結(jié)束后統(tǒng)一補(bǔ)拉。還可以在函數(shù)內(nèi)部加入簡單的指數(shù)退避重試import time import random def fetch_with_retry(user_id, retries3): for attempt in range(retries): try: # 假設(shè)這里是真實的 HTTP 請求 return {user_id: user_id, status: ok} except Exception as exc: wait_time 2 ** attempt random.uniform(0, 1) time.sleep(wait_time) return {user_id: user_id, status: failed}9. Python 多線程面試高頻問題與解答思路結(jié)合前面代碼把面試?yán)锍柕膸讉€問題一并整理。這些問題也適合用來檢驗自己是否真正理解多線程。9.1 Python 的 GIL 是什么GIL 是 CPython 解釋器中的全局解釋器鎖它保證同一時刻只有一個線程在執(zhí)行 Python 字節(jié)碼。多線程無法利用多核 CPU 并行執(zhí)行 CPU 密集型任務(wù)。I/O 密集型任務(wù)不受影響因為等待 I/O 時會釋放 GIL。9.2 GIL 可以被拆除嗎Python 官方多次嘗試移除 GIL但由于 C 擴(kuò)展模塊大量依賴 GIL 做線程安全保護(hù)移除后會影響兼容性和單線程性能。Python 3.13 中提出了自由線程模式但通用生態(tài)尚未完全遷移。實際使用中CPU 密集場景仍然推薦多進(jìn)程。9.3 threading 和 multiprocessing 怎么選線程共享內(nèi)存創(chuàng)建開銷小適合 I/O 密集型進(jìn)程內(nèi)存隔離創(chuàng)建開銷大但能利用多核適合 CPU 密集型。守護(hù)線程和隊列在線程間通信更方便進(jìn)程通信則需要更顯式的機(jī)制。9.4 什么是競態(tài)條件多個線程同時讀寫同一個數(shù)據(jù)資源最終結(jié)果取決于線程調(diào)度順序稱為競態(tài)條件。9.5 Lock 和 RLock 區(qū)別普通Lock只能被 acquire 一次不能重入同一個線程再次 acquire 會死鎖。RLock允許同一個線程多次獲取鎖內(nèi)部有計數(shù)器釋放時也需要對應(yīng)次數(shù)適合遞歸調(diào)用場景。9.6 如何安全地在多線程中共享數(shù)據(jù)優(yōu)先使用queue.Queue這種線程安全容器訪問共享數(shù)據(jù)時必須加鎖。不要依賴“運(yùn)氣”即使某段代碼在小數(shù)據(jù)量測試中沒問題大流量下也可能暴露并發(fā) bug。10. Python 多線程學(xué)習(xí)建議與下一步方向多線程代碼能不能寫對很大程度取決于對共享數(shù)據(jù)和鎖模型的理解。前幾次運(yùn)行出現(xiàn)奇怪結(jié)果非常正常建議刻意做一個實驗把count 1的循環(huán)次數(shù)分別設(shè)為 100 次、1 萬次、100 萬次觀察錯誤概率變化。這個實驗?zāi)苤庇^感覺到數(shù)據(jù)量增大后競態(tài)問題更容易出現(xiàn)。日常開發(fā)中能不用裸線程就不要用裸線程。ThreadPoolExecutor和線程安全隊列幾乎能滿足絕大多數(shù)需求。線程方案如果仍然搞不定高并發(fā) I/O可以再考慮協(xié)程也就是asyncio方案。協(xié)程在單線程里通過事件循環(huán)實現(xiàn)高并發(fā)上下文切換開銷比線程還低適合大量網(wǎng)絡(luò)請求但編程思維和線程明顯不同可以放在后續(xù)學(xué)習(xí)計劃里。100 天精通 Python 系列進(jìn)行到多線程意味著你已經(jīng)具備函數(shù)、模塊、文件類和基礎(chǔ)腳本能力開始進(jìn)入真正的工程并發(fā)領(lǐng)域。今天把threading模塊中不能出 bug 的部分全部掌握代碼全部跑通接下來不管是學(xué)網(wǎng)絡(luò)編程還是爬蟲底氣都會完全不一樣。建議把本文的代碼段單獨(dú)建一個threading_demo目錄保存后續(xù)復(fù)習(xí)時直接運(yùn)行對比實驗結(jié)果。 Python多線程與threading模塊實戰(zhàn)：從線程創(chuàng)建到GIL鎖機(jī)制詳解
在 Python 進(jìn)階路上多線程幾乎是繞不開的一道坎。很多人學(xué)了函數(shù)、類、文件操作之后一旦遇到“下載一批圖片”“同時處理多個請求”“爬蟲抓取多個頁面”這類需求就會發(fā)現(xiàn)單線程程序慢得讓人著急。第 37 天我們來把 Python 的threading模塊徹底講透從一個線程怎么創(chuàng)建到鎖怎么加再到真實場景里到底該不該用多線程全部用代碼過一遍。這篇教程會圍繞幾個核心問題展開Python 里線程和進(jìn)程到底什么關(guān)系、threading模塊怎么用、join和daemon是什么、多線程訪問共享數(shù)據(jù)為什么會出問題、Lock該怎么加以及 GIL 對 Python 多線程到底有什么影響。每個知識點(diǎn)都會給出可以直接運(yùn)行的代碼示例建議讀者順手打開自己的 Python 環(huán)境跟著敲一遍。文末還會附上多線程面試常見問題和工程實踐建議方便對照排查自己的知識盲區(qū)。1. 核心知識點(diǎn)速覽Python 多線程與 threading 模塊主題說明項目定位Python 標(biāo)準(zhǔn)庫多線程編程教程核心模塊threadingPython 內(nèi)置無需 pip 安裝適用 Python 版本Python 3.x3.6 均可推薦 3.8前置知識Python 基礎(chǔ)語法、函數(shù)、類、列表字典操作學(xué)習(xí)目標(biāo)掌握線程創(chuàng)建、啟動、等待、守護(hù)線程、鎖機(jī)制典型應(yīng)用I/O 密集型任務(wù)、網(wǎng)絡(luò)請求、文件讀寫、爬蟲不適合場景CPU 密集型計算受 GIL 限制建議用 multiprocessing運(yùn)行平臺Windows / Linux / macOS 均可運(yùn)行這里先給一個結(jié)論P(yáng)ython 的threading模塊解決的是“并發(fā)”問題不是“并行”問題。它適合讓程序在等待網(wǎng)絡(luò)響應(yīng)、磁盤讀寫時去執(zhí)行別的任務(wù)從而大幅縮短總耗時。2. 使用場景與學(xué)習(xí)邊界什么時候?qū)W多線程什么時候別用多線程不是銀彈學(xué)之前先搞清楚它能解決什么問題也要知道它不適合解決什么問題。2.1 適合用多線程的場景I/O 密集型任務(wù)是 Python 多線程的主場。所謂 I/O 密集型指程序大部分時間都在等待外部設(shè)備返回數(shù)據(jù)CPU 本身反而比較空閑。典型場景包括網(wǎng)絡(luò)爬蟲批量請求網(wǎng)頁每個請求要等服務(wù)器返回。批量下載圖片、文件、視頻。從多個 API 接口拉取數(shù)據(jù)。讀寫大量小文件磁盤 I/O 頻繁但數(shù)據(jù)量不大。數(shù)據(jù)庫查詢量大每次查詢都有網(wǎng)絡(luò)開銷。同時運(yùn)行多個定時任務(wù)或監(jiān)控任務(wù)。在這些場景中一個線程發(fā)起請求后阻塞等待另一個線程立刻被調(diào)度執(zhí)行程序總耗時可縮短到原來的幾分之一甚至幾十分之一。2.2 不適合用多線程的場景CPU 密集型任務(wù)不適合用 Python 多線程。例如大量數(shù)學(xué)計算、圖像像素循環(huán)處理、復(fù)雜算法遞歸、大數(shù)據(jù)排序等。這類任務(wù)需要 CPU 持續(xù)高強(qiáng)度工作而 Python 的 GIL 全局解釋器鎖會讓同一時刻只有一個線程在執(zhí)行 Python 字節(jié)碼多線程反而可能因為線程切換增加額外開銷速度不升反降。CPU 密集場景的正確選擇是multiprocessing進(jìn)程池讓多個進(jìn)程分別占用不同的 CPU 核心。這是第 38 天之后會展開的內(nèi)容本文先不深入。2.3 學(xué)習(xí)路徑建議建議按以下順序?qū)W完本篇文章弄清進(jìn)程與線程的區(qū)別。學(xué)會用threading.Thread創(chuàng)建線程。掌握start()、join()、daemon的使用。理解共享數(shù)據(jù)的線程安全問題。學(xué)會用Lock解決競態(tài)條件。用兩個綜合例程驗證所學(xué)。了解 GIL 對多線程的底層影響。最后做一組面試題自測。3. 環(huán)境準(zhǔn)備Python 多線程編程的前置條件3.1 安裝 Pythonthreading是 Python 標(biāo)準(zhǔn)庫不需要額外安裝。你只需要一個能運(yùn)行 Python 3 的環(huán)境即可。Windows 用戶建議直接到 Python 官網(wǎng)下載安裝包安裝時務(wù)必勾選“Add Python to PATH”。如果安裝后命令行輸入python沒有反應(yīng)說明環(huán)境變量沒有配置好需要手動把 Python 安裝路徑加入 PATH。macOS 用戶建議使用 Homebrew 安裝brew install pythonLinux 用戶多數(shù)發(fā)行版自帶 Python 3可直接使用。沒有的話執(zhí)行sudo apt update sudo apt install python3驗證環(huán)境python --version能輸出 Python 3.x 版本號就說明環(huán)境沒問題。為了代碼編輯體驗更好建議安裝 VS Code 并配置 Python 擴(kuò)展或者直接使用 PyCharm 社區(qū)版。VS Code 里新建.py文件后按CtrlF5即可直接運(yùn)行。3.2 驗證 threading 模塊可用在 Python 交互式環(huán)境或新建腳本中執(zhí)行import threading print(threading.__version__) # Python 3.10 不再暴露版本號會報 AttributeError print(threading.current_thread().name)threading模塊在 Python 3.x 中已經(jīng)非常穩(wěn)定。輸出當(dāng)前線程名稱為MainThread即說明模塊正常工作。3.3 基礎(chǔ)概念準(zhǔn)備寫多線程代碼前先把這幾個核心概念搞清楚。進(jìn)程操作系統(tǒng)分配資源的基本單位。一個程序啟動后就是一個進(jìn)程擁有獨(dú)立的內(nèi)存空間。線程進(jìn)程內(nèi)部的任務(wù)執(zhí)行單元。一個進(jìn)程可以包含多個線程線程之間共享進(jìn)程的內(nèi)存空間。并發(fā)多個任務(wù)交替執(zhí)行看起來像是同時進(jìn)行實際在單核 CPU 上是通過時間片切換完成的。并行多個任務(wù)真的在同一時刻同時執(zhí)行需要多核 CPU 支持。Python 中的 GIL 決定 CPython 解釋器同一時間只允許一個線程執(zhí)行 Python 字節(jié)碼。所以 Python 的多線程更像是一種“并發(fā)”工具用于在等待 I/O 時切換到其他任務(wù)而不是把計算任務(wù)分給多個 CPU 核心并行處理。4. threading 模塊核心用法與代碼實戰(zhàn)4.1 最簡單的線程創(chuàng)建方式直接實例化 Threadthreading.Thread是threading模塊最常用的類。創(chuàng)建線程有兩種方式直接傳入函數(shù)或傳入可調(diào)用對象。第一種方式把普通函數(shù)作為target傳入import threading import time def work(): print(f子線程 {threading.current_thread().name} 開始執(zhí)行) time.sleep(2) print(f子線程 {threading.current_thread().name} 執(zhí)行完畢) print(f主線程 {threading.current_thread().name} 開始) # 創(chuàng)建線程target 指向要執(zhí)行的函數(shù) t threading.Thread(targetwork, nameWorkerThread) # 啟動線程 t.start() print(主線程繼續(xù)往下執(zhí)行)運(yùn)行這段代碼控制臺輸出順序可能類似主線程 MainThread 開始 主線程繼續(xù)往下執(zhí)行 子線程 WorkerThread 開始執(zhí)行 等待 2 秒 子線程 WorkerThread 執(zhí)行完畢注意觀察輸出順序調(diào)用start()之后主線程并不會阻塞等待子線程結(jié)束而是繼續(xù)往下執(zhí)行。子線程什么時候真正被調(diào)度執(zhí)行由操作系統(tǒng)的線程調(diào)度器決定。你可能會發(fā)現(xiàn)WorkerThread 開始執(zhí)行這一行在主線程繼續(xù)往下執(zhí)行之前或之后出現(xiàn)這完全正常。如果你希望主線程等子線程執(zhí)行完畢后再繼續(xù)就需要用到j(luò)oin()import threading import time def work(): print(f子線程 {threading.current_thread().name} 開始執(zhí)行) time.sleep(2) print(f子線程 {threading.current_thread().name} 執(zhí)行完畢) t threading.Thread(targetwork, nameWorkerThread) t.start() # 等待子線程執(zhí)行完畢 t.join() print(主線程在子線程結(jié)束后繼續(xù)執(zhí)行)加上join()后輸出順序變成固定的子線程先完整執(zhí)行完主線程再打最后一行。4.2 自定義線程類繼承 Thread 并重寫 run除了直接傳入函數(shù)還可以通過繼承Thread類并重寫run()方法創(chuàng)建線程。當(dāng)線程調(diào)用start()時內(nèi)部會自動調(diào)用run()方法。這種方式適合邏輯比較復(fù)雜的場景可以把線程相關(guān)的屬性封裝在類中import threading import time class DownloadThread(threading.Thread): def __init__(self, file_name, url): super().__init__() self.file_name file_name self.url url def run(self): print(f[{self.name}] 開始下載 {self.file_name}地址: {self.url}) time.sleep(3) print(f[{self.name}] {self.file_name} 下載完成) if __name__ __main__: tasks [ (圖片1.jpg, https://example.com/img1.jpg), (圖片2.jpg, https://example.com/img2.jpg), (圖片3.jpg, https://example.com/img3.jpg), ] threads [] for name, url in tasks: t DownloadThread(name, url) threads.append(t) t.start() # 等待所有線程執(zhí)行完畢 for t in threads: t.join() print(所有下載任務(wù)已完成)這里self.name是Thread類的內(nèi)置屬性線程未手動命名時系統(tǒng)會自動分配Thread-1、Thread-2這樣的名稱。4.3 線程生命周期與 daemon 守護(hù)線程Python 程序運(yùn)行邏輯中有個重要規(guī)則主線程結(jié)束時如果仍有非守護(hù)線程存活程序會等待所有非守護(hù)線程結(jié)束才退出如果只剩下守護(hù)線程程序會直接結(jié)束。守護(hù)線程通過daemonTrue設(shè)置。典型應(yīng)用是后臺日志收集、心跳檢測、垃圾回收輔助線程等。先看非守護(hù)線程的行為import threading import time def work(): print(子線程開始) time.sleep(3) print(子線程結(jié)束) t threading.Thread(targetwork) t.start() print(主線程即將結(jié)束)輸出順序是子線程開始 主線程即將結(jié)束 程序停留約 3 秒 子線程結(jié)束因為t不是守護(hù)線程主線程執(zhí)行完后必須等t執(zhí)行完程序才會結(jié)束。再看守護(hù)線程import threading import time def work(): print(守護(hù)線程開始) time.sleep(3) print(守護(hù)線程結(jié)束) t threading.Thread(targetwork, daemonTrue) t.start() print(主線程即將結(jié)束)輸出順序可能變成守護(hù)線程開始 主線程即將結(jié)束 程序立即退出守護(hù)線程的結(jié)束語句很可能根本來不及打印因為主線程結(jié)束后程序直接退出不會等守護(hù)線程。實際開發(fā)中守護(hù)線程常用于線程池中的后臺清理任務(wù)?！爱?dāng)主線程結(jié)束守護(hù)線程自動銷毀”這個特性非常方便但要注意守護(hù)線程中的資源清理邏輯不能依賴主線程的返回值。4.4 多線程訪問共享數(shù)據(jù)線程安全問題初現(xiàn)多線程最大的坑不是創(chuàng)建線程而是多個線程同時修改共享數(shù)據(jù)導(dǎo)致結(jié)果錯誤。來看一個經(jīng)典例子import threading count 0 def add(): global count for _ in range(1000000): count 1 def sub(): global count for _ in range(1000000): count - 1 t1 threading.Thread(targetadd) t2 threading.Thread(targetsub) t1.start() t2.start() t1.join() t2.join() print(f最終 count {count})如果兩個線程嚴(yán)格串行執(zhí)行count最終應(yīng)該還是0因為加一百萬次、減一百萬次相互抵消。但實際多次運(yùn)行這段代碼結(jié)果幾乎不可能是 0而是一個正數(shù)或負(fù)數(shù)。這就是典型的競態(tài)條件。原因在于count 1這一行代碼在 Python 底層不是一步完成的它至少包含三條 CPU 指令讀取count當(dāng)前值。執(zhí)行加法運(yùn)算。把新值寫回count。線程 A 執(zhí)行完第 2 步還沒寫回時線程 B 也可能讀取到舊的count值并開始計算。兩個線程同時基于同一個舊值做運(yùn)算最后寫回的新值就相互覆蓋。很多次的更新操作白白丟失最終結(jié)果就偏離了預(yù)期。4.5 Lock 鎖解決競態(tài)條件threading.Lock提供互斥鎖機(jī)制保證同一時刻只有一個線程能進(jìn)入臨界區(qū)。加鎖后的代碼段成為“原子操作”其他線程必須等待鎖被釋放才能進(jìn)入。改進(jìn)后的代碼import threading count 0 lock threading.Lock() def add(): global count for _ in range(1000000): with lock: count 1 def sub(): global count for _ in range(1000000): with lock: count - 1 t1 threading.Thread(targetadd) t2 threading.Thread(targetsub) t1.start() t2.start() t1.join() t2.join() print(f最終 count {count})輸出結(jié)果每次都是穩(wěn)定的0。Lock有兩種使用方式方式一手動調(diào)用acquire()和release()lock.acquire() try: count 1 finally: lock.release()方式二使用with語句推薦的寫法with lock: count 1第二種寫法可以保證即使代碼拋出異常鎖也會被正確釋放。日常開發(fā)中建議只用with方式。關(guān)于鎖還有一些細(xì)節(jié)值得留意。threading.Lock()是可重入的普通鎖嗎答案是普通Lock不可重入。如果同一個線程在未釋放鎖的情況下再次調(diào)用acquire()程序會死鎖。需要可重入鎖的話應(yīng)使用threading.RLock()。另外加鎖范圍越大并發(fā)效率越低。合理做法是只鎖定“真正需要保護(hù)的那幾行代碼”不要在整個循環(huán)外包一層鎖否則兩個線程依然會變成串行執(zhí)行失去多線程的意義。4.6 線程池concurrent.futures 與 ThreadPoolExecutor雖然threading模塊是本文核心但必須提前介紹線程池。實際項目里直接手動創(chuàng)建幾十個裸線程并不是好方案線程的創(chuàng)建和銷毀都有開銷大量線程同時運(yùn)行反而會導(dǎo)致頻繁上下文切換CPU 使用率飆升但任務(wù)完成速度反而下降。Python 推薦用concurrent.futures.ThreadPoolExecutor管理線程from concurrent.futures import ThreadPoolExecutor import time def fetch_data(url): print(f開始請求: {url}) time.sleep(2) return f{url} 的數(shù)據(jù) urls [ https://api.example.com/user/1, https://api.example.com/user/2, https://api.example.com/user/3, https://api.example.com/user/4, https://api.example.com/user/5, ] # 創(chuàng)建最多 3 個線程的線程池 with ThreadPoolExecutor(max_workers3) as executor: futures [executor.submit(fetch_data, url) for url in urls] for future in futures: print(future.result())這段代碼中線程池最多同時運(yùn)行 3 個線程5 個任務(wù)按隊列依次執(zhí)行。with塊結(jié)束時線程池會等待所有任務(wù)完成并自動清理資源。比起手動管理Thread對象的start()和join()簡潔非常多。4.7 線程間通信 Queue多線程之間交換數(shù)據(jù)時手工加鎖管理共享列表很容易出錯。Python 的queue.Queue是線程安全的隊列內(nèi)部自己實現(xiàn)了鎖機(jī)制專門用于多線程生產(chǎn)者-消費(fèi)者模型。import threading import queue import time def producer(q): for i in range(5): item f產(chǎn)品-{i} q.put(item) print(f生產(chǎn)了: {item}) time.sleep(1) def consumer(q): while True: item q.get() if item is None: break print(f消費(fèi)了: {item}) q.task_done() q queue.Queue(maxsize3) threads [ threading.Thread(targetproducer, args(q,)), threading.Thread(targetconsumer, args(q,)), ] for t in threads: t.start() q.join() print(所有任務(wù)完成)生產(chǎn)者線程向隊列添加數(shù)據(jù)消費(fèi)者線程從隊列取出數(shù)據(jù)。Queue內(nèi)部已經(jīng)加鎖多線程同時put或get不會出現(xiàn)數(shù)據(jù)錯亂比手動維護(hù)list加Lock可靠得多。5. 綜合代碼實戰(zhàn)批量下載任務(wù)中的多線程應(yīng)用為了把前面所有知識點(diǎn)串起來這里設(shè)計一個貼近真實業(yè)務(wù)的下載任務(wù)案例。假設(shè)需要在 10 秒內(nèi)從模擬服務(wù)器下載 6 個文件單線程需要約 18 秒用多線程可以把耗時壓縮到約 6 秒。import threading import time import random class MockDownloader: 模擬從網(wǎng)絡(luò)下載文件的類 def __init__(self): self.lock threading.Lock() self.completed 0 def download(self, file_id): # 模擬不同文件大小導(dǎo)致的不同下載時長 cost random.uniform(2, 4) time.sleep(cost) with self.lock: self.completed 1 print(f文件 {file_id} 下載完成耗時 {cost:.2f}s總進(jìn)度 {self.completed}/6) def run(self): threads [] for i in range(1, 7): t threading.Thread(targetself.download, args(i,), namefDownload-{i}) threads.append(t) t.start() for t in threads: t.join() print(f全部下載完成共 {self.completed} 個文件) if __name__ __main__: start time.time() MockDownloader().run() print(f總耗時: {time.time() - start:.2f} 秒)這個案例體現(xiàn)了三個重點(diǎn)多個線程同時發(fā)起“下載”、用threading.current_thread().name或自定義線程名區(qū)分日志、用Lock保護(hù)共享進(jìn)度值避免打印統(tǒng)計錯亂。如果這里不用鎖self.completed 1在極端情況下也會出現(xiàn)計數(shù)丟失的問題。雖然 CPython 對整數(shù)自增有一定優(yōu)化但在高并發(fā)場景下依然不可靠工程上默認(rèn)加鎖是最穩(wěn)妥的。6. 多線程性能觀察與 GIL 影響分析6.1 用實驗證明 I/O 密集型任務(wù)的多線程加速寫一個模擬 I/O 等待的程序?qū)Ρ葐尉€程和多線程耗時import threading import time def io_task(): time.sleep(2) # 單線程執(zhí)行 5 次 start time.time() for _ in range(5): io_task() print(f單線程耗時: {time.time() - start:.2f}s) # 多線程執(zhí)行 5 次 start time.time() threads [] for _ in range(5): t threading.Thread(targetio_task) threads.append(t) t.start() for t in threads: t.join() print(f多線程耗時: {time.time() - start:.2f}s)運(yùn)行結(jié)果通常非常接近單線程耗時: 10.01s 多線程耗時: 2.01s5 個線程同時睡眠等待總耗時不疊加說明多線程成功利用了 I/O 等待時間。6.2 CPU 密集型任務(wù)不要用多線程換一個 CPU 密集型實驗import threading import time def cpu_task(): total 0 for i in range(50000000): total i return total start time.time() for _ in range(4): cpu_task() print(f單線程耗時: {time.time() - start:.2f}s) start time.time() threads [] for _ in range(4): t threading.Thread(targetcpu_task) threads.append(t) t.start() for t in threads: t.join() print(f多線程耗時: {time.time() - start:.2f}s)運(yùn)行后會發(fā)現(xiàn)多線程版本很可能比單線程還慢或者只是持平。原因就是 GIL。CPython 解釋器中任何時刻只允許一個線程執(zhí)行 Python 字節(jié)碼CPU 計算任務(wù)無法利用多核優(yōu)勢。線程越多切換開銷越大性能反而下降。CPU 密集型任務(wù)應(yīng)改用多進(jìn)程。這是官方推薦、社區(qū)公認(rèn)的做法。具體代碼會在后續(xù)文章介紹multiprocessing時詳細(xì)展開本文先記住結(jié)論I/O 密集型用threadingCPU 密集型用multiprocessing。6.3 如何觀察線程運(yùn)行狀態(tài)線程不是越多越好。觀察線程數(shù)量最簡單的方式import threading import time def work(): time.sleep(3) threads [] for i in range(10): t threading.Thread(targetwork) threads.append(t) t.start() print(f當(dāng)前活動線程數(shù): {threading.active_count()}) print(f當(dāng)前線程列表: {threading.enumerate()})threading.active_count()返回當(dāng)前存活線程數(shù)量threading.enumerate()列出所有線程對象。調(diào)試時可以通過打印線程名區(qū)分不同線程的輸出import threading def func(): print(f{threading.current_thread().name} 正在執(zhí)行) t threading.Thread(targetfunc, nameMyWorker) t.start()線程名最好在創(chuàng)建時就指定這樣日志可讀性會好很多。7. 線程安全與常見陷阱排查7.1 直接修改共享 list 或 dict多個線程同時append到同一個list、或同時update同一個dict看起來安全但較大數(shù)據(jù)量下仍可能出現(xiàn)狀態(tài)不一致。不要賭 CPython 的原子性使用鎖保護(hù)共享容器是更嚴(yán)謹(jǐn)?shù)淖龇ā?.2 忘記 join 導(dǎo)致數(shù)據(jù)還沒處理完程序就退出核心工作線程是普通線程主線程忘記join()時主線程雖然不會立即退出但某些邏輯如后續(xù)統(tǒng)計匯總會提前執(zhí)行。建議每個start()都有對應(yīng)的join()不要省略。7.3 死鎖鎖順序不一致多個鎖存在時不同線程按不同順序獲取鎖可能導(dǎo)致互相等待。例如線程 A 持有鎖 1 等待鎖 2線程 B 持有鎖 2 等待鎖 1程序永久卡住。避免死鎖的方法是讓所有線程按相同順序獲取鎖或者盡量用單把鎖控制臨界區(qū)。實際業(yè)務(wù)中要避免鎖嵌套鎖的范圍越小越好。7.4 無限制創(chuàng)建線程導(dǎo)致資源耗盡創(chuàng)建大量線程而不復(fù)用每個線程都占用內(nèi)存和系統(tǒng)資源最終導(dǎo)致 OOM 或線程創(chuàng)建失敗。生產(chǎn)環(huán)境統(tǒng)一用ThreadPoolExecutor(max_workersN)限流不建議直接循環(huán)Thread。7.5daemon線程中的資源清理如果守護(hù)線程持有文件句柄或數(shù)據(jù)庫連接主線程退出時資源可能無法正常釋放。因此守護(hù)線程不要承擔(dān)需要完整清理的任務(wù)寧愿用join()等待或獨(dú)立的進(jìn)程服務(wù)。8. 實戰(zhàn)ThreadPoolExecutor 批量 API 數(shù)據(jù)拉取手動管理十幾個Thread已經(jīng)讓人頭疼真實項目中更推薦直接用線程池。這里給出適合做數(shù)據(jù)采集任務(wù)的通用模板from concurrent.futures import ThreadPoolExecutor, as_completed import time import random def fetch_one(user_id): # 模擬調(diào)用外部 API cost random.uniform(0.5, 2) time.sleep(cost) return {user_id: user_id, status: ok, delay: round(cost, 2)} user_ids list(range(1, 11)) # max_workers 一般設(shè)置為 I/O 型的 5~20 之間需根據(jù)目標(biāo)和API速率調(diào)整 with ThreadPoolExecutor(max_workers5) as executor: future_map {executor.submit(fetch_one, uid): uid for uid in user_ids} for future in as_completed(future_map): uid future_map[future] try: result future.result() print(f用戶 {uid}: {result}) except Exception as exc: print(f用戶 {uid} 請求失敗: {exc})重點(diǎn)看兩點(diǎn)as_completed按任務(wù)完成順序返回結(jié)果不必等待前面的慢任務(wù)可以直接把完成的數(shù)據(jù)寫入文件或數(shù)據(jù)庫。future.result()會拋出任務(wù)內(nèi)部的異常調(diào)用方必須捕獲否則一個任務(wù)出錯可能導(dǎo)致整個主線程中斷。真實采集數(shù)據(jù)時建議配合重試機(jī)制和日志輸出。比如單個請求失敗時先記錄下來全部請求結(jié)束后統(tǒng)一補(bǔ)拉。還可以在函數(shù)內(nèi)部加入簡單的指數(shù)退避重試import time import random def fetch_with_retry(user_id, retries3): for attempt in range(retries): try: # 假設(shè)這里是真實的 HTTP 請求 return {user_id: user_id, status: ok} except Exception as exc: wait_time 2 ** attempt random.uniform(0, 1) time.sleep(wait_time) return {user_id: user_id, status: failed}9. Python 多線程面試高頻問題與解答思路結(jié)合前面代碼把面試?yán)锍柕膸讉€問題一并整理。這些問題也適合用來檢驗自己是否真正理解多線程。9.1 Python 的 GIL 是什么GIL 是 CPython 解釋器中的全局解釋器鎖它保證同一時刻只有一個線程在執(zhí)行 Python 字節(jié)碼。多線程無法利用多核 CPU 并行執(zhí)行 CPU 密集型任務(wù)。I/O 密集型任務(wù)不受影響因為等待 I/O 時會釋放 GIL。9.2 GIL 可以被拆除嗎Python 官方多次嘗試移除 GIL但由于 C 擴(kuò)展模塊大量依賴 GIL 做線程安全保護(hù)移除后會影響兼容性和單線程性能。Python 3.13 中提出了自由線程模式但通用生態(tài)尚未完全遷移。實際使用中CPU 密集場景仍然推薦多進(jìn)程。9.3 threading 和 multiprocessing 怎么選線程共享內(nèi)存創(chuàng)建開銷小適合 I/O 密集型進(jìn)程內(nèi)存隔離創(chuàng)建開銷大但能利用多核適合 CPU 密集型。守護(hù)線程和隊列在線程間通信更方便進(jìn)程通信則需要更顯式的機(jī)制。9.4 什么是競態(tài)條件多個線程同時讀寫同一個數(shù)據(jù)資源最終結(jié)果取決于線程調(diào)度順序稱為競態(tài)條件。9.5 Lock 和 RLock 區(qū)別普通Lock只能被 acquire 一次不能重入同一個線程再次 acquire 會死鎖。RLock允許同一個線程多次獲取鎖內(nèi)部有計數(shù)器釋放時也需要對應(yīng)次數(shù)適合遞歸調(diào)用場景。9.6 如何安全地在多線程中共享數(shù)據(jù)優(yōu)先使用queue.Queue這種線程安全容器訪問共享數(shù)據(jù)時必須加鎖。不要依賴“運(yùn)氣”即使某段代碼在小數(shù)據(jù)量測試中沒問題大流量下也可能暴露并發(fā) bug。10. Python 多線程學(xué)習(xí)建議與下一步方向多線程代碼能不能寫對很大程度取決于對共享數(shù)據(jù)和鎖模型的理解。前幾次運(yùn)行出現(xiàn)奇怪結(jié)果非常正常建議刻意做一個實驗把count 1的循環(huán)次數(shù)分別設(shè)為 100 次、1 萬次、100 萬次觀察錯誤概率變化。這個實驗?zāi)苤庇^感覺到數(shù)據(jù)量增大后競態(tài)問題更容易出現(xiàn)。日常開發(fā)中能不用裸線程就不要用裸線程。ThreadPoolExecutor和線程安全隊列幾乎能滿足絕大多數(shù)需求。線程方案如果仍然搞不定高并發(fā) I/O可以再考慮協(xié)程也就是asyncio方案。協(xié)程在單線程里通過事件循環(huán)實現(xiàn)高并發(fā)上下文切換開銷比線程還低適合大量網(wǎng)絡(luò)請求但編程思維和線程明顯不同可以放在后續(xù)學(xué)習(xí)計劃里。100 天精通 Python 系列進(jìn)行到多線程意味著你已經(jīng)具備函數(shù)、模塊、文件類和基礎(chǔ)腳本能力開始進(jìn)入真正的工程并發(fā)領(lǐng)域。今天把threading模塊中不能出 bug 的部分全部掌握代碼全部跑通接下來不管是學(xué)網(wǎng)絡(luò)編程還是爬蟲底氣都會完全不一樣。建議把本文的代碼段單獨(dú)建一個threading_demo目錄保存后續(xù)復(fù)習(xí)時直接運(yùn)行對比實驗結(jié)果。