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