多线程 threading 基础

程序想同时干多件事——比如同时下载 100 个 URL，或者一边读磁盘一边处理数据——就要用到并发。Python 的并发有三种主流方式：多线程（threading）、多进程（multiprocessing）、异步（asyncio）。这一篇讲多线程：什么时候用它、GIL 是什么、怎么用 threading 和 ThreadPoolExecutor、共享数据的锁、以及常见陷阱。

一、什么是多线程

线程是操作系统调度的最小单位。一个进程里可以有多个线程，它们共享内存，可以并行执行代码（受 GIL 限制，见下节）。

Python 里创建一个线程：

import threading

def worker(name):
    print(f"{name} 开始")
    ...
    print(f"{name} 结束")


t = threading.Thread(target=worker, args=("t1",))
t.start()          # 启动
t.join()           # 等待结束

join() 让主线程等待子线程结束。多个线程：

threads = [threading.Thread(target=worker, args=(f"t{i}",)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

二、GIL：Python 多线程的枷锁

GIL（Global Interpreter Lock） 是 CPython 里的一把全局锁——同一时刻只有一个线程能执行 Python 字节码。

含义：

• CPU 密集型任务（大量计算），多线程不能加速（甚至更慢）；
• IO 密集型任务（读文件、发 HTTP、访问数据库），多线程能显著加速——因为 IO 等待时会释放 GIL。

规则：

• 计算密集 → 用 multiprocessing 或 numpy/cython 等原生库；
• IO 密集 → 用 threading 或 asyncio。

Python 3.13 引入了实验性的 PEP 703（无 GIL） 构建，但目前生产用还是有 GIL 的版本。

三、ThreadPoolExecutor：更 Pythonic

几乎所有场景都推荐用 concurrent.futures 而不是手动 Thread：

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests


def fetch(url):
    r = requests.get(url, timeout=5)
    return url, r.status_code


urls = ["https://httpbin.org/get"] * 20

with ThreadPoolExecutor(max_workers=10) as pool:
    # map：按顺序返回
    for url, status in pool.map(fetch, urls):
        print(url, status)

    # 或者用 submit + as_completed（完成一个处理一个）
    futures = [pool.submit(fetch, url) for url in urls]
    for future in as_completed(futures):
        url, status = future.result()
        print(url, status)

好处：

• 自动管理线程池——不会一次开 100 个线程；
• 用 with 自动清理；
• API 简洁；
• 支持超时、取消、异常传递。

四、Future 对象

submit 返回 Future，代表”未来的结果”：

future = pool.submit(fetch, url)

future.done()             # 是否完成
future.cancel()           # 尝试取消
future.result(timeout=5)  # 阻塞取结果，抛异常也会传上来
future.exception()        # 取异常（如果有）
future.add_done_callback(lambda f: print("完成"))

五、共享数据：Lock

多个线程读写同一份数据要加锁：

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with lock:                   # 加锁块
            counter += 1

threads = [threading.Thread(target=increment) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()

print(counter)     # 应该是 400000

不加锁的话，counter += 1 不是原子的——多线程会覆盖彼此的写。

其他同步原语：

• RLock：可重入锁（同一线程可多次加）；
• Semaphore：信号量（限制并发数）；
• Event：一个线程等另一个 set；
• Condition：条件变量。

六、queue.Queue：线程安全的队列

生产者-消费者模式的最佳搭档：

import queue
import threading
import time


def producer(q):
    for i in range(10):
        print(f"生产 {i}")
        q.put(i)
        time.sleep(0.1)
    q.put(None)      # 结束信号


def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"消费 {item}")


q = queue.Queue()
t1 = threading.Thread(target=producer, args=(q,))
t2 = threading.Thread(target=consumer, args=(q,))
t1.start(); t2.start()
t1.join(); t2.join()

queue.Queue 内部有锁，多线程安全。

七、守护线程

主线程退出时，普通线程会阻塞退出，守护线程会跟着挂：

t = threading.Thread(target=worker, daemon=True)
t.start()
# 主线程退出时，t 也跟着挂（不管是否完成）

适合”后台心跳”、”定时任务”这种”退出就退出”的场景。

八、限制并发数：Semaphore

sem = threading.Semaphore(5)

def limited_task():
    with sem:
        # 同一时刻最多 5 个在这里
        do_stuff()

ThreadPoolExecutor(max_workers=5) 已经内置了这个限制，不用另外加信号量。

九、线程本地存储

想让每个线程有独立的”变量”（比如线程各自的数据库连接）：

local = threading.local()

def worker():
    local.value = threading.current_thread().name    # 每个线程独立
    print(local.value)

十、错误处理

线程里未捕获的异常不会传到主线程：

def bad():
    raise RuntimeError("boom")

t = threading.Thread(target=bad)
t.start()
t.join()          # 什么都不发生，异常"消失"了

用 ThreadPoolExecutor 就好——future.result() 会重抛异常。

十一、什么时候用 threading vs asyncio

场景	推荐
已有大量同步库（requests、mysql-connector）想并发	threading
从零开始的 IO 密集项目	asyncio + httpx
需要与 GUI 交互	threading（GUI 主线程不能阻塞）
处理外部子进程/系统调用	threading 或 subprocess
大量并发连接（10000+）	asyncio

asyncio 的具体讲解见 异步编程 asyncio 入门。

十二、性能测试对比

用 20 个 HTTP 请求做基准：

import time
import requests
from concurrent.futures import ThreadPoolExecutor

urls = ["https://httpbin.org/delay/1"] * 20

# 顺序：约 22 秒
start = time.perf_counter()
for u in urls:
    requests.get(u)
print(f"顺序：{time.perf_counter() - start:.1f}s")

# 多线程：约 1-2 秒
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=20) as pool:
    list(pool.map(requests.get, urls))
print(f"多线程：{time.perf_counter() - start:.1f}s")

多线程能把 IO 密集任务加速 10-100 倍。

十三、一个完整例子：批量下载

# downloader.py
"""并发下载多个 URL 到本地文件。"""

import requests
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging

logger = logging.getLogger(__name__)


def download(url: str, out_dir: Path, timeout: float = 30) -> Path | None:
    """下载单个 URL 到 out_dir，返回本地路径。"""
    try:
        with requests.get(url, timeout=timeout, stream=True) as r:
            r.raise_for_status()
            fname = url.split("/")[-1] or "index.html"
            out = out_dir / fname
            with out.open("wb") as f:
                for chunk in r.iter_content(chunk_size=8192):
                    f.write(chunk)
            logger.info("下载 %s → %s", url, out)
            return out
    except Exception as e:
        logger.error("下载 %s 失败：%s", url, e)
        return None


def batch_download(urls: list[str], out_dir: Path, workers: int = 8) -> list[Path]:
    out_dir.mkdir(parents=True, exist_ok=True)
    results = []

    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(download, u, out_dir): u for u in urls}
        for future in as_completed(futures):
            path = future.result()
            if path:
                results.append(path)

    return results


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO,
                        format="%(asctime)s %(threadName)s %(message)s")

    urls = [
        "https://httpbin.org/uuid",
        "https://httpbin.org/ip",
        "https://httpbin.org/user-agent",
    ] * 5

    downloaded = batch_download(urls, Path("downloads"), workers=5)
    print(f"共下载 {len(downloaded)} 个文件")

十四、常见陷阱

1. CPU 密集用多线程：GIL 让它更慢。用 multiprocessing。
2. 共享数据不加锁：counter += 1 竞态，数据错乱。
3. 在子线程访问 GUI：Qt / Tkinter 都不允许，会崩。
4. 线程数远超 CPU 核心：上下文切换开销大，反而变慢。IO 密集可以开 50-100 个，别开 10000。
5. 忘 join()：主线程先退出，子线程被杀。
6. 子线程异常”消失”：用 ThreadPoolExecutor + future.result()。
7. daemon 线程写文件：主进程退出可能导致文件损坏（写到一半就挂）。

十五、小结与延伸阅读

• 多线程适合IO 密集，不适合CPU 密集（因为 GIL）；
• 优先用 concurrent.futures.ThreadPoolExecutor；
• 共享数据加 threading.Lock；
• 队列用 queue.Queue；
• 每个线程独立变量用 threading.local；
• 异常靠 future.result() 传递；
• 现代新项目建议 asyncio 替代多线程。

延伸阅读：

• threading：https://docs.python.org/zh-cn/3/library/threading.html
• concurrent.futures：https://docs.python.org/zh-cn/3/library/concurrent.futures.html
• GIL 讨论：https://wiki.python.org/moin/GlobalInterpreterLock

下一篇 多进程 multiprocessing 基础 我们讲怎么绕开 GIL 做 CPU 密集任务。