刚开始学Python多线程的时候，我也以为这玩意儿很简单——不就是threading.Thread开几个线程吗？结果连踩三个坑，项目差点崩在线上。今天就把我这两年踩过的坑和优化经验摊开来讲，保证你读完能直接用到实战里。

先看第一个场景：你有个爬虫需要抓取100个网页，单线程跑要10秒，心想“开10个线程不就1秒了？”结果一跑，2.8秒，还时不时卡死。这就是Python的GIL在搞鬼，但更关键的是网络I/O的加速比受限于延迟和带宽，不是简单的线性翻倍。别急，我们一步步解决。

开篇配图：一张GIL锁示意图，显示多线程在CPU密集任务中如何被串行化，红色箭头卡在锁上。

坑一：GIL不是魔鬼，但你得知道怎么绕开它

GIL（全局解释器锁）是Python的“遗传病”，让同一时刻只有一个线程能执行Python字节码。对I/O密集型任务（如爬虫、文件读写），GIL影响不大，因为线程在等待I/O时会释放锁。但对CPU密集型任务（如数值计算），多线程反而比单线程慢。

来看这个错误示范：

“`python
import threading
import time

def cpu_intensive(n):
“””计算斐波那契数列，纯CPU任务”””
a, b = 0, 1
for _ in range(n):
a, b = b, a + b
return a

错误：用多线程跑CPU任务

threads = []
start = time.time()
for i in range(5):
t = threading.Thread(target=cpu_intensive, args=(100000,))
threads.append(t)
t.start()
for t in threads:
t.join()
print(f”多线程耗时: {time.time() – start:.2f}s”) # 输出：约2.1s

对比：单线程顺序跑

start = time.time()
for i in range(5):
cpu_intensive(100000)
print(f”单线程耗时: {time.time() – start:.2f}s”) # 输出：约1.8s
`


看到没？多线程反而更慢。为什么这么写？因为GIL让这5个线程实际上是在抢一个CPU核心，频繁上下文切换反而增加了开销。

解决方案：对于CPU密集型任务，用multiprocessing模块替代threading。每个进程有自己的GIL，能真正利用多核。

`python import multiprocessing as mp import time


def cpu_intensive(n):
    a, b = 0, 1
    for _ in range(n):
        a, b = b, a + b
    return a

if __name__ == "__main__": start = time.time() with mp.Pool(processes=5) as pool: results = pool.map(cpu_intensive, [100000]*5) print(f"多进程耗时: {time.time() - start:.2f}s") # 输出：约0.8s（4核CPU）`


性能从1.8秒降到0.8秒，翻2倍多。记住：线程适合I/O任务，进程适合CPU任务。
坑二：线程安全——别让共享数据变成定时炸弹
另一个坑是共享数据。你写个计数器，多线程一起加，结果最后值不对。看这个经典例子：

`python import threading


counter = 0
def increment():
    global counter
    for _ in range(100000):
        counter += 1  # 在CPython中由于GIL，整数加法实际上是安全的，但为了可移植性和清晰性，仍建议加锁

threads = [] for i in range(10): t = threading.Thread(target=increment) threads.append(t) t.start() for t in threads: t.join() print(f"预期值: 1000000, 实际值: {counter}") # 在CPython中大概率输出1000000，但换其他实现可能出错`

这个设计其实在CPython里大概率能跑对，但别依赖这个——counter += 1在字节码层面是三个步骤：读内存、加一、写回。虽然GIL保证了字节码级别的原子性，但为了代码的可移植性和清晰性，还是老老实实加锁。

解决方案：用threading.Lock加锁，或者用queue.Queue传递数据。官方文档这段文档不够清晰，我直接给你代码：

`python import threading


counter = 0
lock = threading.Lock()
def increment():
    global counter
    for _ in range(100000):
        with lock:  # 自动获取和释放锁
            counter += 1

threads = [] for i in range(10): t = threading.Thread(target=increment) threads.append(t) t.start() for t in threads: t.join() print(f"实际值: {counter}") # 输出：1000000`

用with lock确保同一时刻只有一个线程执行counter += 1。不过加锁会拖慢速度，从0.8秒变1.2秒。另一个技巧是改用threading.local()，每个线程有独立副本，最后再合并，速度更快。


 核心配图：一张线程安全对比图，左侧是数据竞争导致计数器错误，右侧是加锁后正确执行，下附时间对比。
还有个技巧：用线程池管理资源

手动创建和销毁线程开销大。实战中我都是用concurrent.futures.ThreadPoolExecutor，它自动维护线程池，避免造太多线程把系统搞崩。


看这个爬虫案例：

`python import requests import concurrent.futures import time import threading


urls = [f"https://httpbin.org/delay/1" for _ in range(20)]  # 模拟慢API
# 为每个线程创建独立的会话，避免线程安全问题
thread_local = threading.local()
def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session
def fetch_url(url):
    session = get_session()
    response = session.get(url)
    return len(response.text)
单线程
start = time.time()

results = [fetch_url(url) for url in urls]

print(f"单线程耗时: {time.time() - start:.2f}s")  # 约20s
多线程

start = time.time() with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: results = list(executor.map(fetch_url, urls)) print(f"多线程耗时: {time.time() - start:.2f}s") # 约3-4s，受限于网络延迟和带宽`

这样每个线程都有自己的会话，既能复用连接池，又不会踩线程安全的坑。记住：requests.Session`不是线程安全的，别偷懒直接用全局的。