腳本之家服務(wù)器常用軟件

快捷導(dǎo)航

Python threading全景指南分享

更新時間：2025年08月07日 09:09:17 作者：澹溪鶴

Python中線程因GIL限制無法多核并行,但適合I/O密集任務(wù),需用同步工具和線程池管理并發(fā),遵循最佳實踐以提升效率

“并發(fā)不等于并行，但并發(fā)能讓生活更美好。”——《Python 并發(fā)編程實戰(zhàn)》

1. 為什么需要線程

CPU 單核性能已逼近物理極限，要想讓程序在相同時間內(nèi)做更多事，必須“同時”做多件事。

多進程 Process：利用多核并行，資源隔離但開銷大。
協(xié)程 Coroutine：單線程內(nèi)切換，極致 I/O 友好，但無法利用多核。
線程 Thread：介于兩者之間，共享內(nèi)存、切換快，是 I/O 密集型任務(wù)的首選。

在 Python 里，GIL（Global Interpreter Lock）限制了同一進程內(nèi)只能有一條字節(jié)碼在執(zhí)行，進而“弱化”了線程在多核 CPU 上的并行能力。然而：

線程在等待 I/O 時會主動釋放 GIL，因此下載、爬蟲、聊天服務(wù)器等網(wǎng)絡(luò)/磁盤 I/O 場景依舊收益巨大。
對 CPU 密集型任務(wù)，可用 multiprocessing 或 C 擴展繞開 GIL。

一句話：

當(dāng)你想讓程序“邊讀邊寫”“邊收邊發(fā)”“邊阻塞邊響應(yīng)”，就用 threading。

2. 從 0 開始寫線程

2.1 創(chuàng)建線程的兩種姿勢

import threading, time

# 方式一：把函數(shù)塞給 Thread
def worker(n):
    print(f'Worker {n} start')
    time.sleep(1)
    print(f'Worker {n} done')

for i in range(3):
    t = threading.Thread(target=worker, args=(i,))
    t.start()

# 方式二：繼承 Thread 并重寫 run
class MyThread(threading.Thread):
    def __init__(self, n):
        super().__init__()
        self.n = n
    def run(self):
        print(f'MyThread {self.n} start')
        time.sleep(1)
        print(f'MyThread {self.n} done')

MyThread(10).start()

2.2 join：別讓主線程提前跑路

start() 只是告訴操作系統(tǒng)“可以調(diào)度了”，不保證立即執(zhí)行。

threads = [threading.Thread(target=worker, args=(i,)) for i in range(3)]
[t.start() for t in threads]
[t.join() for t in threads]  # 等全部結(jié)束
print('all done')

3. 線程同步：共享變量的“安全帶”

3.1 Lock（互斥鎖）

競爭最激烈的原語，解決“讀寫交叉”問題。

counter = 0
lock = threading.Lock()

def add():
    global counter
    for _ in range(100000):
        with lock:             # 等價于 lock.acquire(); try: ... finally: lock.release()
            counter += 1

threads = [threading.Thread(target=add) for _ in range(2)]
[t.start() for t in threads]
[t.join() for t in threads]
print(counter)   # 200000

沒有 lock 時，大概率得到 <200000 的錯誤結(jié)果。

3.2 RLock（可重入鎖）

同一個線程可以多次 acquire，避免死鎖。

rlock = threading.RLock()
def foo():
    with rlock:
        bar()

def bar():
    with rlock:   # 同一線程，再次獲取成功
        pass

3.3 Condition（條件變量）

經(jīng)典“生產(chǎn)者-消費者”模型：

import random, time
q, MAX = [], 5
cond = threading.Condition()

def producer():
    while True:
        with cond:
            while len(q) == MAX:
                cond.wait()          # 等待隊列有空位
            item = random.randint(1, 100)
            q.append(item)
            print('+', item, q)
            cond.notify()            # 通知消費者
        time.sleep(0.5)

def consumer():
    while True:
        with cond:
            while not q:
                cond.wait()
            item = q.pop(0)
            print('-', item, q)
            cond.notify()
        time.sleep(0.6)

threading.Thread(target=producer, daemon=True).start()
threading.Thread(target=consumer, daemon=True).start()
time.sleep(5)

3.4 Semaphore（信號量）

控制并發(fā)數(shù)量，例如“最多 3 個線程同時下載”。

sem = threading.Semaphore(3)
def downloader(url):
    with sem:
        print('downloading', url)
        time.sleep(2)

3.5 Event（事件）

線程間“發(fā)令槍”機制：

event = threading.Event()

def waiter():
    print('wait...')
    event.wait()          # 阻塞
    print('go!')

threading.Thread(target=waiter).start()
time.sleep(3)
event.set()               # 發(fā)令

3.6 Barrier（柵欄）

N 個線程同時到達某點后再一起繼續(xù)，適合分階段任務(wù)。

barrier = threading.Barrier(3)

def phase(name):
    print(name, 'ready')
    barrier.wait()
    print(name, 'go')

for i in range(3):
    threading.Thread(target=phase, args=(i,)).start()

4. 線程局部變量：ThreadLocal

共享雖好，可有時我們想讓每個線程擁有“私有副本”。

local = threading.local()

def show():
    print(f'{threading.current_thread().name} -> {local.x}')

def task(n):
    local.x = n
    show()

for i in range(3):
    threading.Thread(target=task, args=(i,)).start()

5. 定時器 Timer：延時任務(wù)

def hello():
    print('hello, timer')
threading.Timer(3.0, hello).start()

常用于“超時取消”“心跳包”等場景。

6. 線程池：高并發(fā)下的“資源管家”

頻繁創(chuàng)建/銷毀線程代價高昂，Python 3.2+ 內(nèi)置 concurrent.futures.ThreadPoolExecutor 提供池化能力。

from concurrent.futures import ThreadPoolExecutor
import requests, time

URLS = ['https://baidu.com'] * 20

def fetch(url):
    return requests.get(url).status_code

with ThreadPoolExecutor(max_workers=10) as pool:
    for code in pool.map(fetch, URLS):
        print(code)

max_workers 默認為 min(32, os.cpu_count() + 4)，I/O 密集場景可調(diào)高。
submit + as_completed 組合可實現(xiàn)“誰先完成誰處理”。

7. 調(diào)試與最佳實踐

7.1 死鎖排查

保持加鎖順序一致。
使用 try-lock + 超時。
借助第三方庫 deadlock-debug 或 faulthandler。

7.2 GIL 與性能

CPU 密集：換多進程、Cython、NumPy、multiprocessing。
I/O 密集：放心用線程，瓶頸在網(wǎng)絡(luò)延遲而非 GIL。

7.3 守護線程 daemon

當(dāng)只剩守護線程時，程序直接退出。
常用于后臺心跳、日志寫入，但不要做重要數(shù)據(jù)持久化。

7.4 日志線程名

logging.basicConfig(
    format='%(asctime)s [%(threadName)s] %(message)s',
    level=logging.INFO)

7.5 不要濫用

GUI 程序：UI 線程勿阻塞，耗時操作放后臺線程。
Web 服務(wù)：WSGI 服務(wù)器（uWSGI、gunicorn）已幫你管理進程/線程，業(yè)務(wù)代碼慎用線程。

8. 總結(jié)

維度	線程	進程	協(xié)程
內(nèi)存開銷	低	高	極低
數(shù)據(jù)共享	易	難（需 IPC）	易
切換成本	中	高	極低
適合場景	I/O 密集	CPU 密集	超高并發(fā) I/O
Python 限制	GIL	無	無