今天不聊花哨的技巧，只讲那些年我们亲身趟过的并发天坑，每一个都是实打实的教训，带好小本本，发车！
坑 1：daemon 线程的突然终止
先看这段代码，猜猜输出是什么：
import threadingimport timedef background_task(): time.sleep(1) with open('/tmp/result.txt', 'w') as f: f.write('done')t = threading.Thread(target=background_task, daemon=True)t.start()# 主线程立即结束你以为是 文件被写入？
实际上输出的是 文件可能没有被写入。
daemon 线程在主线程退出时会被强制终止，不会等待它完成。如果 daemon 线程正在执行 I/O 操作（写文件、发网络请求），可能在中途被杀死，导致数据不完整或文件损坏。daemon 线程适合做不重要的后台任务（心跳、监控），不适合做必须完成的工作。
正确写法：
import threadingdef background_task(): with open('/tmp/result.txt', 'w') as f: f.write('done')t = threading.Thread(target=background_task)t.start()t.join() # 等待线程完成踩坑现场：后台线程写日志设成了 daemon，程序异常退出时最后几行日志丢了，排查问题没有线索。
坑 2：多进程中的全局变量
先看这段代码，猜猜输出是什么：
from multiprocessing import Processshared_list = []def worker(val): shared_list.append(val)procs = [Process(target=worker, args=(i,)) for i in range(5)]for p in procs: p.start()for p in procs: p.join()print(shared_list)
你以为是 [0, 1, 2, 3, 4]？
实际上输出的是 []。
多进程不共享内存。每个子进程都有自己独立的内存空间，对全局变量的修改只在子进程内生效，不会反映到主进程。这和多线程完全不同——多线程共享同一进程的内存。要在多进程间共享数据，需要用 multiprocessing.Queue、multiprocessing.Manager 或共享内存。
正确写法：
from multiprocessing import Process, Managerdef worker(shared_list, val): shared_list.append(val)with Manager() as manager: shared_list = manager.list() procs = [Process(target=worker, args=(shared_list, i)) for i in range(5)] for p in procs: p.start() for p in procs: p.join() print(list(shared_list))
踩坑现场：多进程爬虫用全局列表收集结果，跑完发现列表是空的。
坑 3：threading 的 GIL 限制
先看这段代码，猜猜输出是什么：
import threadingimport timedef cpu_work(): total = 0 for i in range(10**7): total += istart = time.time()t1 = threading.Thread(target=cpu_work)t2 = threading.Thread(target=cpu_work)t1.start(); t2.start()t1.join(); t2.join()print(f'多线程: {time.time()-start:.2f}s')start = time.time()cpu_work(); cpu_work()print(f'单线程: {time.time()-start:.2f}s')你以为是 多线程快一倍？
实际上输出的是 多线程和单线程差不多甚至更慢。
CPython 有全局解释器锁（GIL），同一时刻只有一个线程能执行 Python 字节码。对于 CPU 密集型任务，多线程不会带来加速，反而因为线程切换开销可能更慢。多线程在 Python 中只适合 I/O 密集型任务（网络请求、文件读写等）。CPU 密集型任务要用 multiprocessing 或 concurrent.futures.ProcessPoolExecutor。
正确写法：
from multiprocessing import Processimport timedef cpu_work(): total = 0 for i in range(10**7): total += istart = time.time()p1 = Process(target=cpu_work)p2 = Process(target=cpu_work)p1.start(); p2.start()p1.join(); p2.join()print(f'多进程: {time.time()-start:.2f}s')踩坑现场：用多线程做图片处理想加速，发现根本没变快，查资料才知道 GIL 的存在。
坑 4：signal 只能在主线程中使用
报错信息：ValueError: signal only works in main thread
出问题的代码：
import signalimport threadingdef handler(signum, frame): print('caught signal')def setup_in_thread(): signal.signal(signal.SIGTERM, handler)t = threading.Thread(target=setup_in_thread)t.start()t.join()Python 的 signal 模块只能在主线程中使用，在子线程中注册信号处理器会抛出 ValueError。这是因为 Unix 系统信号是进程级别的，发送给主线程处理。如果需要在多线程程序中处理信号，应该在主线程中注册处理器，然后通过 Event 或 Queue 通知其他线程。
修复方法：
import signalimport threadingstop_event = threading.Event()def handler(signum, frame): stop_event.set() # 通知其他线程# 在主线程中注册signal.signal(signal.SIGTERM, handler)def worker(): while not stop_event.is_set(): pass # 工作循环
坑 5：asyncio.run 嵌套调用
报错信息：RuntimeError: asyncio.run() cannot be called from a running event loop
出问题的代码：
import asyncioasync def inner(): return 42async def outer(): return asyncio.run(inner()) # 嵌套调用asyncio.run(outer())
asyncio.run() 会创建一个新的事件循环，但一个线程中只能有一个运行中的事件循环。在已经运行的协程中再调用 asyncio.run() 会报错。在 async 函数中调用其他协程应该用 await，不需要也不能用 asyncio.run()。这个错误在 Jupyter Notebook 中也很常见，因为 Jupyter 已经有一个运行中的事件循环。
修复方法：
import asyncioasync def inner(): return 42async def outer(): return await inner() # 用 awaitresult = asyncio.run(outer())
以上就是今天的 5 个坑。如果你觉得太简单了，说明你运气好还没踩过。
你还踩过哪些让人抓狂的并发大坑？评论区尽情吐槽吧～
来个人儿吧，我快自闭了
下期预告：性能陷阱
