老婆不在家男人玩的Python并发坑：3个高频面试题避坑实录
配置环境就卡半天，明明照着教程敲代码，本地跑得飞起，一到生产环境就崩，这种绝望感谁懂？很多后端开发在应对高频面试题时，喜欢拿 Python 的 threading 或 multiprocessing 举例，觉得能搞定并发就牛了。结果呢？面试时面试官问一句“为什么你的 GIL 没锁死？”，直接哑火。
今天不聊虚的，专门拆解几个让无数开发者栽跟头的 Python 并发陷阱。这些坑，往往就藏在你觉得“肯定没问题”的那几行代码里。无论是应对高频面试题，还是日常写业务代码，避开了这些雷，你的代码稳定性至少提升一个档次。
坑的现象：看似并发，实则串行
很多开发者刚接触 Python 并发，第一反应就是开线程。写个简单的 IO 密集型任务，比如批量请求 API，或者读写文件。代码跑起来，日志打印的时间戳确实是交错的，看起来像是“并发”在执行。
但是，一旦任务稍微重一点，比如涉及到大量的 CPU 计算，或者使用了某些 C 扩展库，性能反而比单线程还慢。更诡异的是，如果你给每个线程加上锁，程序直接卡死，或者吞吐量断崖式下跌。
在面试场景中，这通常是第一道送命题。面试官让你优化一个耗时的脚本，你上来就 threading.Thread，对方微微一笑，问：“你确定你的 Python 进程真的在并行执行 CPU 密集任务吗？”如果你回答不了，后面就不用聊了。
根本原因：GIL 与锁机制的误区
Python 的 GIL（全局解释器锁）是绕不开的话题。很多新人对 GIL 的理解停留在“它是线程锁”这一层面，这是完全错误的。GIL 是解释器级别的锁，它保证同一时刻只有一个线程执行 Python 字节码。
误区一：GIL 让多线程无效。
对于 IO 密集型任务，当线程阻塞在 IO 操作（如网络请求、磁盘读写）时，GIL 会释放。这时候其他线程可以拿到 GIL 继续执行字节码。所以，多线程在 IO 场景下是有效的。但对于 CPU 密集型任务，线程切换的开销（上下文切换、GIL 竞争）远大于执行时间，导致性能不如单线程。
误区二：锁粒度不当。
很多开发者为了“安全”，在循环里加锁，或者把整个函数都包在 with lock: 里。这导致锁的持有时间过长，其他线程长时间阻塞。正确的做法是，锁只保护临界区（即真正需要互斥的那几行代码），粒度越细越好。
误区三：混淆进程与线程。
multiprocessing 模块通过创建子进程来绕过 GIL，实现真正的并行。但进程间通信（IPC）的成本远高于线程共享内存。如果你的任务是 CPU 密集型，用进程；如果是 IO 密集型，用线程或协程。混用会导致资源浪费和调试困难。
正确写法对比：从错误到优雅
下面通过两段代码对比，展示常见的错误写法与推荐写法。假设我们要并发执行 100 个耗时的 HTTP 请求。
错误写法：盲目开线程 + 粗粒度锁
import threading
import time
import requests# 全局锁，粒度太粗
lock = threading.Lock()
results = []def fetch(url):# 错误点1：在IO等待期间也持有了锁，导致串行化with lock:try:resp = requests.get(url, timeout=5)# 错误点2：数据组装也在锁内，增加了锁持有时间data = resp.json()results.append(data)except Exception as e:print(fError: {e})# 错误点3：异常处理在锁内，可能导致锁未正确释放（虽然with会处理，但逻辑不清晰）def main():threads = []for i in range(100):t = threading.Thread(target=fetch, args=(fhttp://httpbin.org/delay/1,))threads.append(t)t.start()for t in threads:t.join()# 假设这里处理resultsprint(fTotal: {len(results)})if __name__ == __main__:start = time.time()main()print(fTime: {time.time() - start})问题解析：with lock 包裹了整个 fetch 函数，包括 requests.get。这意味着第一个线程发出请求后，必须等到响应回来，其他线程才能开始发请求。这完全丧失了并发意义，变成了串行执行。
锁的粒度太大，包含了网络 IO 和数据解析，极大地降低了并发度。
没有使用线程池，直接创建 100 个线程，可能导致系统资源耗尽（文件描述符、内存）。正确写法：线程池 + 细粒度锁 + 无锁收集
import concurrent.futures
import requests
import timeresults = []
# 错误点修正：不再需要全局锁来保护列表添加，因为我们将使用线程池的结果收集机制
# 或者，如果必须用列表，可以加锁，但只在添加时加锁def fetch(url):# 正确点1：不在IO操作外加锁try:resp = requests.get(url, timeout=5)data = resp.json()# 正确点2：如果results是共享列表，这里需要细粒度锁# 但在concurrent.futures中，我们通常直接返回结果，由主线程收集return dataexcept Exception as e:# 正确点3：异常在函数内部处理，不影响其他线程print(fError: {e})return Nonedef main():urls = [fhttp://httpbin.org/delay/1 for _ in range(100)]# 正确点4：使用线程池，限制最大线程数，避免资源耗尽with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:# 正确点5：提交任务，获取Future对象future_to_url = {executor.submit(fetch, url): url for url in urls}# 正确点6：使用as_completed，按完成顺序处理结果，而不是等待所有线程joinfor future in concurrent.futures.as_completed(future_to_url):data = future.result()if data is not None:results.append(data)print(fTotal: {len(results)})if __name__ == __main__:start = time.time()main()print(fTime: {time.time() - start})优势解析：线程池复用：ThreadPoolExecutor 自动管理线程生命周期，避免频繁创建销毁线程的开销。max_workers=20 限制了并发数，既保证了并发度，又控制了资源消耗。
细粒度控制：IO 操作完全无锁并行执行。只有当结果需要写入共享结构时，才考虑同步机制（此处由主线程统一收集，天然线程安全）。
非阻塞等待：as_completed 允许主线程在任何一个任务完成后立即处理，而不是傻等到所有线程都结束。这在处理长尾任务时特别有用。
异常隔离：单个请求失败不会阻塞其他请求，也不会导致整个程序崩溃。复现与修复代码：进阶避坑技巧
除了上述基础用法，还有几个高频面试中容易踩的坑，这里给出复现和修复方案。
坑1：死锁
现象：程序卡死，没有任何输出，CPU 占用率极低。
原因：两个线程分别持有了对方需要的锁，并且都在等待对方释放。
错误代码：
import threadinglock_a = threading.Lock()
lock_b = threading.Lock()def thread_1():with lock_a:time.sleep(0.1) # 模拟耗时with lock_b:passdef thread_2():with lock_b:time.sleep(0.1)with lock_a:pass# 启动线程，必然死锁
t1 = threading.Thread(target=thread_1)
t2 = threading.Thread(target=thread_2)
t1.start()
t2.start()
t1.join()
t2.join()修复方案：固定锁顺序：所有线程必须按照相同的顺序获取锁（例如，先 lock_a 再 lock_b）。
使用超时：lock.acquire(timeout=5)，如果获取失败则释放已持有的锁并重试或退出。
避免嵌套锁：重新设计数据结构，尽量使用无锁数据结构（如 queue.Queue）或细粒度锁。坑2：竞态条件（Race Condition）
现象：数据不一致，比如计数器结果不对，或者列表出现重复/缺失元素。
原因：多个线程同时读写共享变量，且操作不可原子性。
错误代码：
import threadingcounter = 0
lock = threading.Lock()def increment():global counter# 错误：读-改-写不是原子操作temp = countertime.sleep(0.001) # 模拟耗时，增加竞态窗口counter = temp + 1# 启动10个线程，每个线程执行1000次
threads = []
for _ in range(10):t = threading.Thread(target=lambda: [increment() for _ in range(1000)])threads.append(t)t.start()for t in threads:t.join()print(fExpected: 10000, Actual: {counter}) # 结果通常远小于10000修复方案：加锁保护临界区：def increment_safe():global counterwith lock:counter += 1 # 在锁保护下执行原子操作使用原子操作：Python 中 list.append 是原子的（受 GIL 保护），但 counter += 1 不是。可以使用 itertools.count 或专门的原子计数器库。
使用 threading.Lock 或 threading.RLock：确保所有对共享状态的修改都在锁保护下进行。坑3：内存泄漏
现象：程序运行越久，内存占用越高，最终 OOM。
原因：线程或进程未正确回收，或者闭包中引用了大对象。
常见场景：创建了线程但未 join，且线程持有大对象引用。
使用了 weakref 不当，导致对象被提前回收或无法回收。
在循环中创建线程，但未清理。修复建议：及时回收：使用 with 语句管理线程池或进程池，确保资源释放。
避免闭包引用：在回调函数中，避免引用外部的大对象。如果需要，使用 weakref。
监控内存：使用 tracemalloc 或 memory_profiler 工具定位内存泄漏点。规避建议：从代码到架构选择正确的并发模型：IO 密集型：asyncio（协程） ThreadPoolExecutor（线程） multiprocessing（进程）。
CPU 密集型：multiprocessing Cython/PyPy 单线程优化。
混合类型：拆分为 IO 部分和 CPU 部分，分别使用合适的模型。最小化共享状态：尽量让线程或进程处理独立的数据集。
使用消息队列（如 queue.Queue）进行通信，而不是共享内存。使用成熟的库：不要自己实现线程池或进程池，使用 concurrent.futures。
对于高并发场景，考虑使用 asyncio + aiohttp 等异步库。测试与监控：编写压力测试，模拟高并发场景。
监控线程数、CPU 使用率、内存占用等指标。
使用日志记录关键并发事件，便于排查问题。阅读官方文档：Python 官方文档中关于 threading 和 concurrent.futures 的章节非常详细，包含了大量的最佳实践和注意事项。不要只靠博客和教程，官方文档才是最终依据。并发编程是后端开发的必修课，也是高频面试题的重灾区。避坑的关键在于理解 GIL 的本质、锁的粒度以及并发模型的选择。不要迷信多线程，要根据实际场景选择合适的工具。
你在并发编程中遇到过哪些奇葩的坑？或者在高频面试题中被问到过哪些让你头疼的并发问题？还有什么不懂的？评论区留言挨个回。 老婆不在家男人玩的Python并发坑：3个高频面试题避坑实录
老婆不在家男人玩的Python并发坑：3个高频面试题避坑实录
配置环境就卡半天，明明照着教程敲代码，本地跑得飞起，一到生产环境就崩，这种绝望感谁懂？很多后端开发在应对高频面试题时，喜欢拿 Python 的 threading 或 multiprocessing 举例，觉得能搞定并发就牛了。结果呢？面试时面试官问一句“为什么你的 GIL 没锁死？”，直接哑火。
今天不聊虚的，专门拆解几个让无数开发者栽跟头的 Python 并发陷阱。这些坑，往往就藏在你觉得“肯定没问题”的那几行代码里。无论是应对高频面试题，还是日常写业务代码，避开了这些雷，你的代码稳定性至少提升一个档次。
坑的现象：看似并发，实则串行
很多开发者刚接触 Python 并发，第一反应就是开线程。写个简单的 IO 密集型任务，比如批量请求 API，或者读写文件。代码跑起来，日志打印的时间戳确实是交错的，看起来像是“并发”在执行。
但是，一旦任务稍微重一点，比如涉及到大量的 CPU 计算，或者使用了某些 C 扩展库，性能反而比单线程还慢。更诡异的是，如果你给每个线程加上锁，程序直接卡死，或者吞吐量断崖式下跌。
在面试场景中，这通常是第一道送命题。面试官让你优化一个耗时的脚本，你上来就 threading.Thread，对方微微一笑，问：“你确定你的 Python 进程真的在并行执行 CPU 密集任务吗？”如果你回答不了，后面就不用聊了。
根本原因：GIL 与锁机制的误区
Python 的 GIL（全局解释器锁）是绕不开的话题。很多新人对 GIL 的理解停留在“它是线程锁”这一层面，这是完全错误的。GIL 是解释器级别的锁，它保证同一时刻只有一个线程执行 Python 字节码。
误区一：GIL 让多线程无效。
对于 IO 密集型任务，当线程阻塞在 IO 操作（如网络请求、磁盘读写）时，GIL 会释放。这时候其他线程可以拿到 GIL 继续执行字节码。所以，多线程在 IO 场景下是有效的。但对于 CPU 密集型任务，线程切换的开销（上下文切换、GIL 竞争）远大于执行时间，导致性能不如单线程。
误区二：锁粒度不当。
很多开发者为了“安全”，在循环里加锁，或者把整个函数都包在 with lock: 里。这导致锁的持有时间过长，其他线程长时间阻塞。正确的做法是，锁只保护临界区（即真正需要互斥的那几行代码），粒度越细越好。
误区三：混淆进程与线程。
multiprocessing 模块通过创建子进程来绕过 GIL，实现真正的并行。但进程间通信（IPC）的成本远高于线程共享内存。如果你的任务是 CPU 密集型，用进程；如果是 IO 密集型，用线程或协程。混用会导致资源浪费和调试困难。
正确写法对比：从错误到优雅
下面通过两段代码对比，展示常见的错误写法与推荐写法。假设我们要并发执行 100 个耗时的 HTTP 请求。
错误写法：盲目开线程 + 粗粒度锁
import threading
import time
import requests# 全局锁，粒度太粗
lock = threading.Lock()
results = []def fetch(url):# 错误点1：在IO等待期间也持有了锁，导致串行化with lock:try:resp = requests.get(url, timeout=5)# 错误点2：数据组装也在锁内，增加了锁持有时间data = resp.json()results.append(data)except Exception as e:print(fError: {e})# 错误点3：异常处理在锁内，可能导致锁未正确释放（虽然with会处理，但逻辑不清晰）def main():threads = []for i in range(100):t = threading.Thread(target=fetch, args=(fhttp://httpbin.org/delay/1,))threads.append(t)t.start()for t in threads:t.join()# 假设这里处理resultsprint(fTotal: {len(results)})if __name__ == __main__:start = time.time()main()print(fTime: {time.time() - start})问题解析：with lock 包裹了整个 fetch 函数，包括 requests.get。这意味着第一个线程发出请求后，必须等到响应回来，其他线程才能开始发请求。这完全丧失了并发意义，变成了串行执行。
锁的粒度太大，包含了网络 IO 和数据解析，极大地降低了并发度。
没有使用线程池，直接创建 100 个线程，可能导致系统资源耗尽（文件描述符、内存）。正确写法：线程池 + 细粒度锁 + 无锁收集
import concurrent.futures
import requests
import timeresults = []
# 错误点修正：不再需要全局锁来保护列表添加，因为我们将使用线程池的结果收集机制
# 或者，如果必须用列表，可以加锁，但只在添加时加锁def fetch(url):# 正确点1：不在IO操作外加锁try:resp = requests.get(url, timeout=5)data = resp.json()# 正确点2：如果results是共享列表，这里需要细粒度锁# 但在concurrent.futures中，我们通常直接返回结果，由主线程收集return dataexcept Exception as e:# 正确点3：异常在函数内部处理，不影响其他线程print(fError: {e})return Nonedef main():urls = [fhttp://httpbin.org/delay/1 for _ in range(100)]# 正确点4：使用线程池，限制最大线程数，避免资源耗尽with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:# 正确点5：提交任务，获取Future对象future_to_url = {executor.submit(fetch, url): url for url in urls}# 正确点6：使用as_completed，按完成顺序处理结果，而不是等待所有线程joinfor future in concurrent.futures.as_completed(future_to_url):data = future.result()if data is not None:results.append(data)print(fTotal: {len(results)})if __name__ == __main__:start = time.time()main()print(fTime: {time.time() - start})优势解析：线程池复用：ThreadPoolExecutor 自动管理线程生命周期，避免频繁创建销毁线程的开销。max_workers=20 限制了并发数，既保证了并发度，又控制了资源消耗。
细粒度控制：IO 操作完全无锁并行执行。只有当结果需要写入共享结构时，才考虑同步机制（此处由主线程统一收集，天然线程安全）。
非阻塞等待：as_completed 允许主线程在任何一个任务完成后立即处理，而不是傻等到所有线程都结束。这在处理长尾任务时特别有用。
异常隔离：单个请求失败不会阻塞其他请求，也不会导致整个程序崩溃。复现与修复代码：进阶避坑技巧
除了上述基础用法，还有几个高频面试中容易踩的坑，这里给出复现和修复方案。
坑1：死锁
现象：程序卡死，没有任何输出，CPU 占用率极低。
原因：两个线程分别持有了对方需要的锁，并且都在等待对方释放。
错误代码：
import threadinglock_a = threading.Lock()
lock_b = threading.Lock()def thread_1():with lock_a:time.sleep(0.1) # 模拟耗时with lock_b:passdef thread_2():with lock_b:time.sleep(0.1)with lock_a:pass# 启动线程，必然死锁
t1 = threading.Thread(target=thread_1)
t2 = threading.Thread(target=thread_2)
t1.start()
t2.start()
t1.join()
t2.join()修复方案：固定锁顺序：所有线程必须按照相同的顺序获取锁（例如，先 lock_a 再 lock_b）。
使用超时：lock.acquire(timeout=5)，如果获取失败则释放已持有的锁并重试或退出。
避免嵌套锁：重新设计数据结构，尽量使用无锁数据结构（如 queue.Queue）或细粒度锁。坑2：竞态条件（Race Condition）
现象：数据不一致，比如计数器结果不对，或者列表出现重复/缺失元素。
原因：多个线程同时读写共享变量，且操作不可原子性。
错误代码：
import threadingcounter = 0
lock = threading.Lock()def increment():global counter# 错误：读-改-写不是原子操作temp = countertime.sleep(0.001) # 模拟耗时，增加竞态窗口counter = temp + 1# 启动10个线程，每个线程执行1000次
threads = []
for _ in range(10):t = threading.Thread(target=lambda: [increment() for _ in range(1000)])threads.append(t)t.start()for t in threads:t.join()print(fExpected: 10000, Actual: {counter}) # 结果通常远小于10000修复方案：加锁保护临界区：def increment_safe():global counterwith lock:counter += 1 # 在锁保护下执行原子操作使用原子操作：Python 中 list.append 是原子的（受 GIL 保护），但 counter += 1 不是。可以使用 itertools.count 或专门的原子计数器库。
使用 threading.Lock 或 threading.RLock：确保所有对共享状态的修改都在锁保护下进行。坑3：内存泄漏
现象：程序运行越久，内存占用越高，最终 OOM。
原因：线程或进程未正确回收，或者闭包中引用了大对象。
常见场景：创建了线程但未 join，且线程持有大对象引用。
使用了 weakref 不当，导致对象被提前回收或无法回收。
在循环中创建线程，但未清理。修复建议：及时回收：使用 with 语句管理线程池或进程池，确保资源释放。
避免闭包引用：在回调函数中，避免引用外部的大对象。如果需要，使用 weakref。
监控内存：使用 tracemalloc 或 memory_profiler 工具定位内存泄漏点。规避建议：从代码到架构选择正确的并发模型：IO 密集型：asyncio（协程） ThreadPoolExecutor（线程） multiprocessing（进程）。
CPU 密集型：multiprocessing Cython/PyPy 单线程优化。
混合类型：拆分为 IO 部分和 CPU 部分，分别使用合适的模型。最小化共享状态：尽量让线程或进程处理独立的数据集。
使用消息队列（如 queue.Queue）进行通信，而不是共享内存。使用成熟的库：不要自己实现线程池或进程池，使用 concurrent.futures。
对于高并发场景，考虑使用 asyncio + aiohttp 等异步库。测试与监控：编写压力测试，模拟高并发场景。
监控线程数、CPU 使用率、内存占用等指标。
使用日志记录关键并发事件，便于排查问题。阅读官方文档：Python 官方文档中关于 threading 和 concurrent.futures 的章节非常详细，包含了大量的最佳实践和注意事项。不要只靠博客和教程，官方文档才是最终依据。并发编程是后端开发的必修课，也是高频面试题的重灾区。避坑的关键在于理解 GIL 的本质、锁的粒度以及并发模型的选择。不要迷信多线程，要根据实际场景选择合适的工具。
你在并发编程中遇到过哪些奇葩的坑？或者在高频面试题中被问到过哪些让你头疼的并发问题？还有什么不懂的？评论区留言挨个回。