写Python并发相关的文章我其实挺抗拒一上来就贴threading模块的 API 文档。这年头资料太容易查了真正让人卡壳的是线程到底在 Python 里是怎么跑起来的——为什么加了线程 CPU 没跑满为什么我的共享变量老是被改乱为什么有人一说 Python 线程就摇头这些问题的根源全在线程的实现细节里而不是那几个start()、join()方法上。所以我这篇不打算念 API 手册而是把 CPython 解释器层面、操作系统调度层面、以及你写的那几行业务代码之间到底发生了什么一条线捋清楚。你不需要是 C 语言专家我也尽量不用一堆晦涩的术语糊脸但读完你应该能回答这三个问题Python 线程是什么、它能干什么、它死在哪儿。顺带我把自己实测过的数据、踩过的坑、排查死锁的土办法也一并放进来这些是文档里不会写的。1. 从线程是什么到Python 线程是什么聊 Python 的线程实现之前得先把概念推倒重建一遍。很多初学者把线程理解成代码里可以并行跑的几段逻辑这个理解不算错但在 Python 里它会严重误导你。1.1 线程的本质操作系统的调度单位线程是操作系统内核里的一个调度实体。每个线程有自己的栈、寄存器上下文、程序计数器操作系统负责把它们排队、切换、分配 CPU 时间片。你用 C 或者 Java 创建线程本质上就是调用操作系统提供的pthread_createLinux/macOS或CreateThreadWindows这样的接口。Python 里的threading.Thread并不特殊——它没有自己发明一套绿色线程或者协程来假装自己是线程。CPython 解释器在遇到threading.Thread()时底层照样会去调用操作系统原生的线程创建接口。也就是说你在 Python 里开的每一个线程在操作系统眼里都是真实存在的、可以被调度器分配到一个 CPU 核心上去执行的任务。这一点很多人搞混觉得Python 线程是假的、是模拟出来的。不是的。它是真的纯的货真价实的原生系统线程该有的创建开销、上下文切换开销、栈空间占用一个都跑不掉。1.2 threading 模块与_thread的封装关系Python 的线程体系分两层底层是_thread模块它是对操作系统线程 API 的轻量封装提供_thread.start_new_thread()这种最朴素的创建方式。上层是你日常用的threading模块它在_thread之上做了大量面向对象的封装Thread类、Lock、RLock、Condition、Event、Semaphore、Barrier等等。实际用的时候你几乎不会直接碰_thread但你要知道这一层封装存在。threading.Thread创建线程对象时并不是start()才真正创建线程——它是在start()调用时才通过_thread去内核创建原生线程然后让这个线程去执行你传入的run()方法。import threading import time def worker(name): print(f线程 {name} 运行在 native_id{threading.get_native_id()}) time.sleep(1) t threading.Thread(targetworker, args(A,)) t.start() t.join()注意threading.get_native_id()这个函数返回的是操作系统线程 ID不是 Python 层的ident。你可以用系统命令比如 Linux 的ps -eLf或者htop看到这个线程真实存在于内核的线程列表里。如果你心理上一直觉得 Python 线程是模拟的做个这个小实验立刻就会有实感。1.3 线程创建过程的完整链路画一条链路的话是这样的threading.Thread(...).start() └─ 检查线程是否已启动、是否 daemon └─ _thread.start_new_thread(thread_entry, (self,)) └─ CPython 内部调用 PyThread_start_new_thread └─ pthread_createLinux/ CreateThreadWindows └─ 内核分配栈空间、初始化 TCB、加入调度队列这条链路里藏着几个值得知道的开销点栈空间分配每个线程默认栈大小在 Linux 上通常是 8MB 虚拟内存。注意是虚拟内存物理内存按需分配所以不是说你开 1000 个线程就吃掉 8GB 物理内存但虚拟地址空间确实会占着。创建和销毁的系统调用开销一次线程创建大概几微秒到几十微秒不等但频繁创建销毁会在内核态和用户态之间来回切积累起来的开销相当可观。上下文切换成本线程数超过 CPU 核心数之后切换线程涉及保存/恢复寄存器、栈指针、缓存失效等操作。这个成本在你线程数量失控时会急剧显现。理解这条链路后面很多性能直觉就能建立了。比如有人喜欢在循环里无脑Thread(targetxxx).start()每个请求都新开线程系统负载一大就开始卡顿——那就是因为创建/销毁线程本身就是一个不小的成本而且会触发频繁的系统调用。这个问题的标准解法不是优化线程创建时间而是直接用线程池把创建和销毁的开销摊销掉。2. 绕不开的 GIL一把锁的代价与妥协只要谈 Python 线程就不能绕过 GILGlobal Interpreter Lock全局解释器锁。很多刚接触的人觉得这是 Python 的缺陷但我的观点是这是一个历史设计决策有代价但远非一无是处。理解它的机制你才知道什么时候该用线程什么时候该换进程或者换写法。2.1 GIL 到底是什么为什么还留着GIL 是 CPython 解释器内部的一把互斥锁。它的作用是在同一个进程内任意时刻只允许一个线程执行 Python 字节码。注意这句话有两层意思它锁的是执行 Python 字节码这件事。它在同一个进程内生效——多进程模式下每个进程有自己独立的解释器和独立的 GIL互不干扰。为什么需要这把锁因为 CPython 的内存管理引用计数不是线程安全的。如果不加锁两个线程同时操作同一个对象的引用计数就会导致对象被提前回收或者内存泄漏这种 bug 排查起来是灾难级的。与其给每个对象都加细粒度的锁不如整个解释器共用一把大锁实现简单、稳定、开发迭代快。这是 CPython 在执行性能和开发效率、稳定性之间做的一个明确取舍。Jython、IronPython 等实现没有 GIL但它们的生态跟进程度远不如 CPython。所以现实就是你用的 Python几乎肯定是 CPython几乎肯定有 GIL。2.2 GIL 的切换机制不是每个字节码都切换Python 3.2 之前GIL 的切换基于字节码指令计数执行够一定数量的指令就切换。3.2 之后改成了基于时间的机制默认情况下一个线程持有 GIL 执行sys.getswitchinterval()秒默认 5 毫秒之后会主动释放 GIL让其他线程有机会运行。这个机制有个重要细节释放 GIL 不等于马上换线程。释放之后操作系统要唤醒另一个线程并让它抢到 GIL这个过程涉及线程调度、锁竞争实际开销并不小。另外很多阻塞型 I/O 操作会主动释放 GIL。比如time.sleep()、socket.recv()、文件读写、requests.get()这类操作在等待期间GIL 是放开的。这就是为什么 I/O 密集型任务用线程能获得很好的并发提升——线程在等 I/O 的时候并不占着 GIL其他线程可以继续干活。纯 CPU 计算的代码就完全是另一个故事了。比如一段没有任何 I/O 的while循环它不会主动释放 GIL只能等 5 毫秒时间片到期被强制切换。这种情况下线程多了不仅不加速反而会因为频繁的 GIL 竞争和上下文切换变得更慢。来个简单验证实验直接在命令行跑python - EOF import sys print(sys.getswitchinterval()) EOF我的机器上输出是0.005也就是 5ms。你可以把这个值改小比如sys.setswitchinterval(0.001)会看到线程切换变得更频繁极端情况下 CPU 密集型任务表现得比单线程还差。2.3 GIL 不等于线程安全这是新手最容易踩的坑。既然 GIL 保证同一时刻只有一个线程执行字节码那是不是就不需要加锁了完全不是。GIL 保护的是解释器层面的指令安全不是你业务逻辑的原子性。一条 Python 语句可能对应好几条字节码中间完全可能发生线程切换。最经典的例子就是这个import threading counter 0 def increment(): global counter for _ in range(500000): counter 1 threads [threading.Thread(targetincrement) for _ in range(2)] for t in threads: t.start() for t in threads: t.join() print(counter) # 你会发现不是 1000000counter 1这行代码在字节码层面至少需要三条指令读取counter的值、把值加一、把新值写回。这三条指令之间 GIL 可能已经切换了好几次。两个线程同时读到了旧的counter值各自加一后写回结果相当于只加了一次。你说它是并发 bug也行说它是线程不安全也行总之结果是错的。所以正确姿势永远是共享可变数据记得加锁或者用queue.Queue、concurrent.futures这些线程安全的容器。不要指望 GIL 帮你兜底。2.4 自由线程时代的开端2023 年底CPython 社区正式接受了 PEP 703Making the Global Interpreter Lock OptionalPython 3.13 开始提供实验性的自由线程构建free-threaded build。这个构建模式下GIL 可以被禁用Python 线程可以真正并行使用多核 CPU。不过要泼点冷水目前这只是实验特性生态里大量第三方 C 扩展最典型的就是很多科学计算库还没适配需要显式声明支持才能保证线程安全。我自己的态度是关注进展但生产环境别急着上。等生态适配成熟了Python 线程的并行能力会有质的提升但那是未来的事眼下你把 GIL 的行为摸透才是能立刻用上的技能。3. 实战对比线程在 CPU 密集和 I/O 密集下的真实表现理论说再多不如跑个测试直观。我用自己的机器8 核 CPULinuxPython 3.11跑了一组简单的基准测试分别测单线程、多线程在不同类型任务下的表现。结果非常有参考价值。3.1 CPU 密集型线程真的会拖后腿CPU 密集型任务就是纯计算比如大量数学运算、复杂的算法循环、数据转换。我写了一个计算斐波那契数列的函数来模拟import threading import time def fib(n): if n 2: return n return fib(n - 1) fib(n - 2) def run_single(): start time.perf_counter() fib(32) print(f单线程耗时: {time.perf_counter() - start:.4f}s) def run_multi(): start time.perf_counter() threads [threading.Thread(targetfib, args(32,)) for _ in range(4)] for t in threads: t.start() for t in threads: t.join() print(f4 线程耗时: {time.perf_counter() - start:.4f}s) run_single() run_multi()我跑了三次取平均结果大概是任务方式耗时秒单线程计算一次0.514 线程各计算一次并行2.15你没看错4 个线程并行执行 4 个同样的计算任务总耗时是单线程的 4 倍还多。如果完全并行理想情况下应该接近单线程耗时可惜 GIL 的存在让纯计算任务根本无法并行。线程之间轮流抢 GIL加上切换开销结果比串行还差。这就是Python 线程不适合 CPU 密集型任务的原因不是线程本身有问题是 GIL 在这条路上设了卡。这种场景的正确工具是multiprocessing多进程或者把计算挪到 numpy、C 扩展等能释放 GIL 的库里去。3.2 I/O 密集型线程的主场I/O 密集型任务的典型特征是有大量等待比如网络请求、文件读写、数据库查询。等待期间不占 CPU也不占 GIL。我用一个模拟网络请求的函数来测假设每次请求要等 0.5 秒import threading import time def io_task(): time.sleep(0.5) # 模拟一次网络 I/O 等待 def run_single(): start time.perf_counter() for _ in range(10): io_task() print(f串行耗时: {time.perf_counter() - start:.4f}s) def run_multi(): start time.perf_counter() threads [threading.Thread(targetio_task) for _ in range(10)] for t in threads: t.start() for t in threads: t.join() print(f10 线程耗时: {time.perf_counter() - start:.4f}s) run_single() run_multi()结果任务方式耗时秒串行执行 10 次5.0310 个线程并行执行0.51这个提升非常直观同样是 10 个请求串行要等 5 秒线程并发只需要 0.5 秒——所有等待全部重叠了。在这类场景下线程就是神器。爬虫批量抓取、批量调用 API、大量文件读写用线程池可以轻松把吞吐拉高一个数量级。3.3 线程池 vs 手动开线程既然手动开线程能跑为什么还要线程池自己Thread().start()每次都要创建线程、最后还得join()收尸代码丑不说线程数量一多、任务一杂管理起来就头大。线程池解决了三个问题复用线程避免频繁创建/销毁的开销。限制并发数量防止无脑把机器跑死。统一管理任务提交、结果获取和异常处理。concurrent.futures.ThreadPoolExecutor是标准库里的首选。我最常用的写法是配合as_completed或mapfrom concurrent.futures import ThreadPoolExecutor, as_completed import time def fetch(url): time.sleep(0.5) return fok: {url} urls [fhttps://example.com/page/{i} for i in range(20)] with ThreadPoolExecutor(max_workers10) as executor: future_map {executor.submit(fetch, url): url for url in urls} for future in as_completed(future_map): url future_map[future] try: result future.result() print(f{url} - {result}) except Exception as e: print(f{url} - 出错: {e})with块结束时会自动等待所有任务完成并关闭线程池不用手动 join省心很多。max_workers怎么选我一般看场景I/O 密集型可以多开比如 10~20 个甚至更多因为线程大部分时间在等CPU 密集型别多开线程数按 CPU 核心数来就行。这不是理论教条是我实测出来再往上加线程吞吐量不涨反跌的经验值。4. 线程间的数据共享与同步线程之所以比多进程轻一个重要原因就是线程之间天然共享进程的内存空间。你可以直接在一个线程里修改全局变量另一个线程立刻能看到。这把双刃剑用好了效率极高用不好就是各种诡异的 bug。4.1 变性竞争一个计数器引发的血案前面 GIL 那节已经展示过counter 1的竞争问题。这里再往深处挖一点竞争条件race condition不只是数学算错还可能导致死锁、数据损坏、甚至程序崩溃。解决竞争最直接的工具是threading.Lockimport threading counter 0 lock threading.Lock() def increment(): global counter for _ in range(500000): with lock: counter 1 threads [threading.Thread(targetincrement) for _ in range(2)] for t in threads: t.start() for t in threads: t.join() print(counter) # 正确输出 1000000with lock:块保证同一时刻只有一个线程能进入临界区。注意Lock是不可重入的同一个线程在未释放的情况下再次acquire()会直接死锁。如果业务逻辑里出现持锁方法调用另一个持锁方法的情况就得用RLock可重入锁它允许同一个线程多次获取。4.2 不只是 Lock其他同步原语怎么选Lock是最基础的但实际场景里还有很多需求它不是最优解。我整理了一张选型表原语适用场景特点Lock保护共享资源的互斥访问不可重入简单粗暴RLock同一线程可能多次获取锁可重入避免自我死锁Event一个线程通知另一个线程可以开始了简单的事件标志Condition生产者-消费者模式需要等待特定条件可以配合wait/notify使用Semaphore限制同时访问资源的线程数控制并发额度Barrier多个线程等待彼此到达某一点同步屏障queue.Queue线程间安全传递数据内部自带锁推荐优先使用我的经验是能用queue.Queue解决的就别自己手动加锁。Queue内部实现了线程安全的入队和出队还支持阻塞等待、超时控制极难用错。手动锁的代码越复杂死锁风险越高。比如生产者-消费者模式我最简单的写法就是import queue import threading import time q queue.Queue(maxsize5) def producer(): for i in range(10): q.put(i) print(f生产: {i}) time.sleep(0.2) def consumer(): while True: item q.get() if item is None: # 哨兵值通知退出 break print(f消费: {item}) time.sleep(0.3) threads [ threading.Thread(targetproducer), threading.Thread(targetconsumer), ] for t in threads: t.start() q.join() # 等队列中所有任务被处理完 # 生产结束后放哨兵注意q.join()和task_done()的配合使用。Queue内部维护一个未完成任务计数每次get()后调用task_done()才把计数减一join()才会返回。很多新手漏了task_done()导致join()永远阻塞。4.3 死锁的典型场景与排查思路死锁的经典场景是互相持有对方需要的锁。A 线程持有锁 1 等锁 2B 线程持有锁 2 等锁 1两边互不相让程序永远卡死。import threading import time lock_a threading.Lock() lock_b threading.Lock() def task_a(): with lock_a: time.sleep(0.1) # 故意让两个线程撞上 with lock_b: print(task_a done) def task_b(): with lock_b: time.sleep(0.1) with lock_a: print(task_b done) t1 threading.Thread(targettask_a) t2 threading.Thread(targettask_b) t1.start() t2.start() t1.join() t2.join() # 程序卡死在这里这个代码跑起来就是教科书级的死锁。排查手段我推荐三个层次faulthandler.dump_traceback_later()在死锁场景下定时 dump 所有线程的堆栈。这是标准库内置的不需要装额外工具我最常用。py-spy dump --pid PID第三方工具可以附着到运行中的进程打印所有线程当前的堆栈对排查线上死锁特别有效。缺点是 Linux 下需要一定权限。pdb手动断点适合本地复现断点打在可疑的acquire附近手工看线程状态。写代码时预防死锁的几条经验锁的获取顺序保持一致。多个锁嵌套时全局统一顺序比如都先拿 A 再拿 B。尽量缩小临界区锁内不做耗时操作尤其不要做 I/O网络请求、文件写。用acquire(timeout...)给锁加超时获取不到就放弃或者重试至少不会无限卡死。4.4 线程本地数据不想共享就别共享有些数据每个线程应该各有一份不需要也不应该共享比如每个线程独立的数据库连接、会话、随机数种子。这时候用threading.local()再合适不过。import threading local_data threading.local() def worker(name): local_data.thread_name name # 每个线程独立设置互不干扰 print(f{threading.current_thread().name}: {local_data.thread_name}) threading.Thread(targetworker, args(线程A,)).start() threading.Thread(targetworker, args(线程B,)).start()local_data在不同线程里访问到的值是隔离的类似一个线程维度的字典。在很多并发框架包括 web 框架里用它来保存请求级别的上下文比全局变量安全得多。5. 几个容易踩的生产级细节纸上谈兵差不多了聊几个我在真实项目里踩过的坑。这些坑不大但每一个都足以让你排查大半天。5.1 守护线程daemon的退出陷阱threading.Thread(daemonTrue)表示这是一个守护线程。主线程退出时守护线程会被强制终止不执行清理、不触发finally、不做任何善后。区分需求后台任务如果允许主程序退出就放弃可以设daemonTrue但如果这个线程负责写日志、刷新缓存、优雅关闭连接那绝对不能设守护必须join()等它干完。我最常犯的错是写个轮询线程想着反正是后台任务就设了daemonTrue结果程序退出时还会收到 SIGTERM导致日志文件里的最后几条记录丢了。改成非守护线程后主流程join()等待这个线程正常退出问题就解决了。5.2 异常在子线程里是沉默的子线程抛出异常主线程完全无感知——程序不会报错主线程该干嘛干嘛。这个坑特别隐蔽因为本地测试时线程启动的异常可能还能在控制台看到一旦上了服务框架子线程异常直接进日志黑洞。我的习惯是在线程入口包一层try/except手动把异常传给主线程或写到专门的错误队列import threading import queue error_q queue.Queue() def safe_worker(): try: # 实际业务逻辑 raise ValueError(演示异常) except Exception as e: error_q.put(e) t threading.Thread(targetsafe_worker) t.start() t.join() if not error_q.empty(): err error_q.get() print(f子线程捕获到异常: {err})如果有第三方代码在线程里执行一旦不确定是否静默吞异常最好统一走这个模式。5.3 线程创建的爆炸式增长防不胜防每来一个请求就开一个线程这种代码我在很多项目的早期版本里都见过。请求量一大线程数飙升到几千甚至上万每个线程还要占 8MB 虚拟栈空间、消耗上下文切换资源服务直接假死。上文提到的ThreadPoolExecutor是标准解但还有一个细节要注意如果任务队列里积压了大量任务线程池再大也是杯水车薪。此时应该做的是限流和削峰比如用信号量限制同时处理的任务数或者直接上消息队列做缓冲。线程池解决的是并发执行问题不解决任务洪水问题这两个概念千万别混。5.4 文件写入乱序别让多个线程抢同一个句柄多线程同时往同一个文件里写日志写出来的内容会乱成一团——不是内容丢了而是交错穿插。因为文件对象本身不是线程安全的。解决方案有三类所有写文件操作集中到一个线程其他线程把日志内容丢进queue.Queue由这个线程统一写。用日志库logging.handlers里内置的锁机制logging模块的FileHandler本身是线程安全的。真需要高性能并发写考虑每个线程独立文件异步合并。我强烈推荐第一种或者第二种。手动加锁写文件也能跑但会在写文件这种低耗时操作上引入锁竞争吞吐量在日志量大时下降明显。6. 线程与协程、多进程怎么配合使用线程不是 Python 并发世界的唯一答案。协程asyncio、多进程multiprocessing各有各的适用场景但实际项目里它们经常不是互斥关系而是协同关系。6.1 选型逻辑看你的瓶颈在哪瓶颈推荐方案核心原因网络 I/O、文件 I/O 等待协程或线程等待期让出执行权复用率高CPU 密集计算多进程绕开 GIL真正利用多核两者混合多进程 线程/协程各司其职计算和等待分离线程和协程的选择上我的经验是如果项目从零开始、主要就是高并发 I/O 场景而且团队熟悉异步编程asyncio的吞吐量上限更高因为它没有线程的上下文切换和内核态开销。但如果业务逻辑里有大量同步的第三方库调用比如很多数据库驱动、老版本 requests硬上asyncio反而痛苦线程是更务实的选择。这里有个实用技巧在asyncio事件循环里调用同步阻塞库时用loop.run_in_executor(None, func, args)把它丢到线程池里执行这样既保持了异步架构又不会阻塞事件循环。这是我处理异步框架 同步库兼容问题的最常用招数。import asyncio import time def blocking_query(x): time.sleep(1) return x * 2 async def main(): loop asyncio.get_running_loop() # 把阻塞操作丢到线程池避免阻塞事件循环 result await loop.run_in_executor(None, blocking_query, 21) print(result) asyncio.run(main())6.2 线程 进程的混合模式我做过一个写爬虫的项目目标站点多、单次请求耗时高、返回数据还需要做大量 JSON 解析和清洗。如果纯用线程解析阶段抢 GIL 拖慢整体如果纯用进程进程之间的数据传递成本高。最后是混合模式multiprocessing开 4 个进程每个进程内部用ThreadPoolExecutor跑 10 个线程做请求和初步解析解析结果通过multiprocessing.Queue汇总到主进程做最终处理。这样请求等待和 CPU 解析都得到了最大的资源利用。这个方案实现复杂度更高但吞吐量比纯线程版本高了将近一倍。如果你的任务同时具备I/O 等待 CPU 计算两种特征可以考虑这个方向。7. 实测总结线程并发的数量红线与性能参考最后给点硬核参考。我粗略统计了近期多个项目在真实负载下的线程模型数据谈不上严谨压测但给你一个不太会跑偏的感觉机器配置线程场景实测参考4 核 8G 云主机线程池 I/O 密集任务max_workers20~50工作良好16 核 32G 物理机线程池 I/O 密集任务max_workers100~200可达吞吐顶峰8 核桌面 PCCPU 密集任务 线程线程数超过 2 就开始明显拖慢任何配置手动每请求一线程线程数 500 后上下文切换开销显著延迟飙升注意这些数字会因具体任务特征浮动不能当作万能公式抄。我的建议是I/O 密集场景从CPU 核数 × 10起步跑一轮真实流量看 CPU 延迟曲线再逐步调高千万别一上来拍脑袋定个大数。7.1 线程数的调优方法一个笨但可靠的方法是做个简单的阶梯压力测试固定任务量比如 1000 个模拟请求分别设置max_workers10/20/50/100/200。记录总耗时和每个请求的平均延迟。画个趋势总耗时先降后升拐点就是你的最优并发数。以我的经验I/O 密集任务的最优并发数通常远大于 CPU 核数因为线程都在等网络包但超过一定阈值后线程切换和 GIL 争抢成本会盖过并发收益。那个天花板在哪只有实测才能找到。7.2 线程与 I/O 事件循环的对比感受用同样的任务量分别用线程池和asyncio跑我的经验数据是纯 I/O 场景下asyncio的开销大概是线程的 1/51/3内存占用和 CPU 占用都更省。但代码复杂度会明显上升尤其是在处理限流、重试、超时这类业务逻辑时。我的建议很实际团队熟哪个用哪个别为了高性能盲目换异步框架。并发编程的第一目标是正确和可维护性能提升建立在正确的架构之上。用线程把并发跑通、保证正确性再考虑要不要换成异步来优化这个迭代路径远比一上来就上asyncio稳。8. 经验之谈线程调试的几把土武器说了这么多理论和数据最后留点私货——我在排查线程问题时的几个土办法。工具未必高大上但确实帮我解决过不少头疼的问题。8.1 日志里带线程名和线程 ID多线程环境打日志不带线程信息等于白打。我固定用threading.current_thread().name和threading.get_native_id()拼进日志前缀这样一出问题翻日志就能按线程维度筛。import threading import logging logging.basicConfig( levellogging.INFO, format%(asctime)s [%(threadName)s] [tid%(threadid)s] %(message)s, )threadName可以直接用tid需要额外一个LogRecord过滤器传入或者直接在日志消息里手动拼。日志配合线程维度排查并发问题至少能筛掉一半的障碍。8.2 用 faulthandler 抓卡死现场faulthandler是标准库但知道的人不算多。它可以在程序卡死的状态下定期把每个线程的 Python 堆栈打印到 stderr 或者文件里。import faulthandler import threading import time faulthandler.dump_traceback_later(5, repeatTrue) # 5 秒后开始 dump持续重复 def deadlock_task(): lock threading.Lock() lock.acquire() lock.acquire() # 自我死锁 threading.Thread(targetdeadlock_task).start() time.sleep(30)运行时你能看到每个线程当前停在哪个函数哪一行立刻定位卡点在哪儿。如果你在云服务器上排查问题这招比 GDB 好使太多不需要额外装任何东西。8.3 用py-spy看线上进程py-spy是我最常用的第三方调试工具比gdb方便不需要程序以调试模式运行。py-spy dump --pid PID直接打印那个进程里所有线程的 Python 堆栈。遇到线上卡死attach 上去瞬间就知道线程卡在锁、网、IO 的哪一层。注意py-spy在容器里运行时需要一定的权限配置特定 Linux 发行版可能还要设置ptrace_scope。但为了线上排查问题值得提前配好。8.4 复现竞态条件的小技巧竞态条件往往不稳定复现今天跑得好好的明天就炸。我的技巧是在关键临界区前后故意加一点随机阻塞把竞态的概率窗口人为拉大让 bug 更容易暴露。import random import time def flaky_worker(): time.sleep(random.uniform(0.001, 0.01)) # 进入临界区...这个技巧没法用来修复 bug但用来验证是不是并发问题一测一个准。真正修复还是得靠正确的同步原语和设计。最后再分享一个小技巧写并发代码永远先从最朴素的正确方案开始比如加一把大锁保证逻辑正确然后再打磨性能比如缩小临界区、换读写锁、改无锁结构。我见过太多人一上来就追求无锁设计结果花了大量时间调 bug最后还不如一把锁来得快。并发编程里正确性先于性能这句话什么时候都不过时。 Python线程深度解析：从GIL到并发实战经验
写Python并发相关的文章我其实挺抗拒一上来就贴threading模块的 API 文档。这年头资料太容易查了真正让人卡壳的是线程到底在 Python 里是怎么跑起来的——为什么加了线程 CPU 没跑满为什么我的共享变量老是被改乱为什么有人一说 Python 线程就摇头这些问题的根源全在线程的实现细节里而不是那几个start()、join()方法上。所以我这篇不打算念 API 手册而是把 CPython 解释器层面、操作系统调度层面、以及你写的那几行业务代码之间到底发生了什么一条线捋清楚。你不需要是 C 语言专家我也尽量不用一堆晦涩的术语糊脸但读完你应该能回答这三个问题Python 线程是什么、它能干什么、它死在哪儿。顺带我把自己实测过的数据、踩过的坑、排查死锁的土办法也一并放进来这些是文档里不会写的。1. 从线程是什么到Python 线程是什么聊 Python 的线程实现之前得先把概念推倒重建一遍。很多初学者把线程理解成代码里可以并行跑的几段逻辑这个理解不算错但在 Python 里它会严重误导你。1.1 线程的本质操作系统的调度单位线程是操作系统内核里的一个调度实体。每个线程有自己的栈、寄存器上下文、程序计数器操作系统负责把它们排队、切换、分配 CPU 时间片。你用 C 或者 Java 创建线程本质上就是调用操作系统提供的pthread_createLinux/macOS或CreateThreadWindows这样的接口。Python 里的threading.Thread并不特殊——它没有自己发明一套绿色线程或者协程来假装自己是线程。CPython 解释器在遇到threading.Thread()时底层照样会去调用操作系统原生的线程创建接口。也就是说你在 Python 里开的每一个线程在操作系统眼里都是真实存在的、可以被调度器分配到一个 CPU 核心上去执行的任务。这一点很多人搞混觉得Python 线程是假的、是模拟出来的。不是的。它是真的纯的货真价实的原生系统线程该有的创建开销、上下文切换开销、栈空间占用一个都跑不掉。1.2 threading 模块与_thread的封装关系Python 的线程体系分两层底层是_thread模块它是对操作系统线程 API 的轻量封装提供_thread.start_new_thread()这种最朴素的创建方式。上层是你日常用的threading模块它在_thread之上做了大量面向对象的封装Thread类、Lock、RLock、Condition、Event、Semaphore、Barrier等等。实际用的时候你几乎不会直接碰_thread但你要知道这一层封装存在。threading.Thread创建线程对象时并不是start()才真正创建线程——它是在start()调用时才通过_thread去内核创建原生线程然后让这个线程去执行你传入的run()方法。import threading import time def worker(name): print(f线程 {name} 运行在 native_id{threading.get_native_id()}) time.sleep(1) t threading.Thread(targetworker, args(A,)) t.start() t.join()注意threading.get_native_id()这个函数返回的是操作系统线程 ID不是 Python 层的ident。你可以用系统命令比如 Linux 的ps -eLf或者htop看到这个线程真实存在于内核的线程列表里。如果你心理上一直觉得 Python 线程是模拟的做个这个小实验立刻就会有实感。1.3 线程创建过程的完整链路画一条链路的话是这样的threading.Thread(...).start() └─ 检查线程是否已启动、是否 daemon └─ _thread.start_new_thread(thread_entry, (self,)) └─ CPython 内部调用 PyThread_start_new_thread └─ pthread_createLinux/ CreateThreadWindows └─ 内核分配栈空间、初始化 TCB、加入调度队列这条链路里藏着几个值得知道的开销点栈空间分配每个线程默认栈大小在 Linux 上通常是 8MB 虚拟内存。注意是虚拟内存物理内存按需分配所以不是说你开 1000 个线程就吃掉 8GB 物理内存但虚拟地址空间确实会占着。创建和销毁的系统调用开销一次线程创建大概几微秒到几十微秒不等但频繁创建销毁会在内核态和用户态之间来回切积累起来的开销相当可观。上下文切换成本线程数超过 CPU 核心数之后切换线程涉及保存/恢复寄存器、栈指针、缓存失效等操作。这个成本在你线程数量失控时会急剧显现。理解这条链路后面很多性能直觉就能建立了。比如有人喜欢在循环里无脑Thread(targetxxx).start()每个请求都新开线程系统负载一大就开始卡顿——那就是因为创建/销毁线程本身就是一个不小的成本而且会触发频繁的系统调用。这个问题的标准解法不是优化线程创建时间而是直接用线程池把创建和销毁的开销摊销掉。2. 绕不开的 GIL一把锁的代价与妥协只要谈 Python 线程就不能绕过 GILGlobal Interpreter Lock全局解释器锁。很多刚接触的人觉得这是 Python 的缺陷但我的观点是这是一个历史设计决策有代价但远非一无是处。理解它的机制你才知道什么时候该用线程什么时候该换进程或者换写法。2.1 GIL 到底是什么为什么还留着GIL 是 CPython 解释器内部的一把互斥锁。它的作用是在同一个进程内任意时刻只允许一个线程执行 Python 字节码。注意这句话有两层意思它锁的是执行 Python 字节码这件事。它在同一个进程内生效——多进程模式下每个进程有自己独立的解释器和独立的 GIL互不干扰。为什么需要这把锁因为 CPython 的内存管理引用计数不是线程安全的。如果不加锁两个线程同时操作同一个对象的引用计数就会导致对象被提前回收或者内存泄漏这种 bug 排查起来是灾难级的。与其给每个对象都加细粒度的锁不如整个解释器共用一把大锁实现简单、稳定、开发迭代快。这是 CPython 在执行性能和开发效率、稳定性之间做的一个明确取舍。Jython、IronPython 等实现没有 GIL但它们的生态跟进程度远不如 CPython。所以现实就是你用的 Python几乎肯定是 CPython几乎肯定有 GIL。2.2 GIL 的切换机制不是每个字节码都切换Python 3.2 之前GIL 的切换基于字节码指令计数执行够一定数量的指令就切换。3.2 之后改成了基于时间的机制默认情况下一个线程持有 GIL 执行sys.getswitchinterval()秒默认 5 毫秒之后会主动释放 GIL让其他线程有机会运行。这个机制有个重要细节释放 GIL 不等于马上换线程。释放之后操作系统要唤醒另一个线程并让它抢到 GIL这个过程涉及线程调度、锁竞争实际开销并不小。另外很多阻塞型 I/O 操作会主动释放 GIL。比如time.sleep()、socket.recv()、文件读写、requests.get()这类操作在等待期间GIL 是放开的。这就是为什么 I/O 密集型任务用线程能获得很好的并发提升——线程在等 I/O 的时候并不占着 GIL其他线程可以继续干活。纯 CPU 计算的代码就完全是另一个故事了。比如一段没有任何 I/O 的while循环它不会主动释放 GIL只能等 5 毫秒时间片到期被强制切换。这种情况下线程多了不仅不加速反而会因为频繁的 GIL 竞争和上下文切换变得更慢。来个简单验证实验直接在命令行跑python - EOF import sys print(sys.getswitchinterval()) EOF我的机器上输出是0.005也就是 5ms。你可以把这个值改小比如sys.setswitchinterval(0.001)会看到线程切换变得更频繁极端情况下 CPU 密集型任务表现得比单线程还差。2.3 GIL 不等于线程安全这是新手最容易踩的坑。既然 GIL 保证同一时刻只有一个线程执行字节码那是不是就不需要加锁了完全不是。GIL 保护的是解释器层面的指令安全不是你业务逻辑的原子性。一条 Python 语句可能对应好几条字节码中间完全可能发生线程切换。最经典的例子就是这个import threading counter 0 def increment(): global counter for _ in range(500000): counter 1 threads [threading.Thread(targetincrement) for _ in range(2)] for t in threads: t.start() for t in threads: t.join() print(counter) # 你会发现不是 1000000counter 1这行代码在字节码层面至少需要三条指令读取counter的值、把值加一、把新值写回。这三条指令之间 GIL 可能已经切换了好几次。两个线程同时读到了旧的counter值各自加一后写回结果相当于只加了一次。你说它是并发 bug也行说它是线程不安全也行总之结果是错的。所以正确姿势永远是共享可变数据记得加锁或者用queue.Queue、concurrent.futures这些线程安全的容器。不要指望 GIL 帮你兜底。2.4 自由线程时代的开端2023 年底CPython 社区正式接受了 PEP 703Making the Global Interpreter Lock OptionalPython 3.13 开始提供实验性的自由线程构建free-threaded build。这个构建模式下GIL 可以被禁用Python 线程可以真正并行使用多核 CPU。不过要泼点冷水目前这只是实验特性生态里大量第三方 C 扩展最典型的就是很多科学计算库还没适配需要显式声明支持才能保证线程安全。我自己的态度是关注进展但生产环境别急着上。等生态适配成熟了Python 线程的并行能力会有质的提升但那是未来的事眼下你把 GIL 的行为摸透才是能立刻用上的技能。3. 实战对比线程在 CPU 密集和 I/O 密集下的真实表现理论说再多不如跑个测试直观。我用自己的机器8 核 CPULinuxPython 3.11跑了一组简单的基准测试分别测单线程、多线程在不同类型任务下的表现。结果非常有参考价值。3.1 CPU 密集型线程真的会拖后腿CPU 密集型任务就是纯计算比如大量数学运算、复杂的算法循环、数据转换。我写了一个计算斐波那契数列的函数来模拟import threading import time def fib(n): if n 2: return n return fib(n - 1) fib(n - 2) def run_single(): start time.perf_counter() fib(32) print(f单线程耗时: {time.perf_counter() - start:.4f}s) def run_multi(): start time.perf_counter() threads [threading.Thread(targetfib, args(32,)) for _ in range(4)] for t in threads: t.start() for t in threads: t.join() print(f4 线程耗时: {time.perf_counter() - start:.4f}s) run_single() run_multi()我跑了三次取平均结果大概是任务方式耗时秒单线程计算一次0.514 线程各计算一次并行2.15你没看错4 个线程并行执行 4 个同样的计算任务总耗时是单线程的 4 倍还多。如果完全并行理想情况下应该接近单线程耗时可惜 GIL 的存在让纯计算任务根本无法并行。线程之间轮流抢 GIL加上切换开销结果比串行还差。这就是Python 线程不适合 CPU 密集型任务的原因不是线程本身有问题是 GIL 在这条路上设了卡。这种场景的正确工具是multiprocessing多进程或者把计算挪到 numpy、C 扩展等能释放 GIL 的库里去。3.2 I/O 密集型线程的主场I/O 密集型任务的典型特征是有大量等待比如网络请求、文件读写、数据库查询。等待期间不占 CPU也不占 GIL。我用一个模拟网络请求的函数来测假设每次请求要等 0.5 秒import threading import time def io_task(): time.sleep(0.5) # 模拟一次网络 I/O 等待 def run_single(): start time.perf_counter() for _ in range(10): io_task() print(f串行耗时: {time.perf_counter() - start:.4f}s) def run_multi(): start time.perf_counter() threads [threading.Thread(targetio_task) for _ in range(10)] for t in threads: t.start() for t in threads: t.join() print(f10 线程耗时: {time.perf_counter() - start:.4f}s) run_single() run_multi()结果任务方式耗时秒串行执行 10 次5.0310 个线程并行执行0.51这个提升非常直观同样是 10 个请求串行要等 5 秒线程并发只需要 0.5 秒——所有等待全部重叠了。在这类场景下线程就是神器。爬虫批量抓取、批量调用 API、大量文件读写用线程池可以轻松把吞吐拉高一个数量级。3.3 线程池 vs 手动开线程既然手动开线程能跑为什么还要线程池自己Thread().start()每次都要创建线程、最后还得join()收尸代码丑不说线程数量一多、任务一杂管理起来就头大。线程池解决了三个问题复用线程避免频繁创建/销毁的开销。限制并发数量防止无脑把机器跑死。统一管理任务提交、结果获取和异常处理。concurrent.futures.ThreadPoolExecutor是标准库里的首选。我最常用的写法是配合as_completed或mapfrom concurrent.futures import ThreadPoolExecutor, as_completed import time def fetch(url): time.sleep(0.5) return fok: {url} urls [fhttps://example.com/page/{i} for i in range(20)] with ThreadPoolExecutor(max_workers10) as executor: future_map {executor.submit(fetch, url): url for url in urls} for future in as_completed(future_map): url future_map[future] try: result future.result() print(f{url} - {result}) except Exception as e: print(f{url} - 出错: {e})with块结束时会自动等待所有任务完成并关闭线程池不用手动 join省心很多。max_workers怎么选我一般看场景I/O 密集型可以多开比如 10~20 个甚至更多因为线程大部分时间在等CPU 密集型别多开线程数按 CPU 核心数来就行。这不是理论教条是我实测出来再往上加线程吞吐量不涨反跌的经验值。4. 线程间的数据共享与同步线程之所以比多进程轻一个重要原因就是线程之间天然共享进程的内存空间。你可以直接在一个线程里修改全局变量另一个线程立刻能看到。这把双刃剑用好了效率极高用不好就是各种诡异的 bug。4.1 变性竞争一个计数器引发的血案前面 GIL 那节已经展示过counter 1的竞争问题。这里再往深处挖一点竞争条件race condition不只是数学算错还可能导致死锁、数据损坏、甚至程序崩溃。解决竞争最直接的工具是threading.Lockimport threading counter 0 lock threading.Lock() def increment(): global counter for _ in range(500000): with lock: counter 1 threads [threading.Thread(targetincrement) for _ in range(2)] for t in threads: t.start() for t in threads: t.join() print(counter) # 正确输出 1000000with lock:块保证同一时刻只有一个线程能进入临界区。注意Lock是不可重入的同一个线程在未释放的情况下再次acquire()会直接死锁。如果业务逻辑里出现持锁方法调用另一个持锁方法的情况就得用RLock可重入锁它允许同一个线程多次获取。4.2 不只是 Lock其他同步原语怎么选Lock是最基础的但实际场景里还有很多需求它不是最优解。我整理了一张选型表原语适用场景特点Lock保护共享资源的互斥访问不可重入简单粗暴RLock同一线程可能多次获取锁可重入避免自我死锁Event一个线程通知另一个线程可以开始了简单的事件标志Condition生产者-消费者模式需要等待特定条件可以配合wait/notify使用Semaphore限制同时访问资源的线程数控制并发额度Barrier多个线程等待彼此到达某一点同步屏障queue.Queue线程间安全传递数据内部自带锁推荐优先使用我的经验是能用queue.Queue解决的就别自己手动加锁。Queue内部实现了线程安全的入队和出队还支持阻塞等待、超时控制极难用错。手动锁的代码越复杂死锁风险越高。比如生产者-消费者模式我最简单的写法就是import queue import threading import time q queue.Queue(maxsize5) def producer(): for i in range(10): q.put(i) print(f生产: {i}) time.sleep(0.2) def consumer(): while True: item q.get() if item is None: # 哨兵值通知退出 break print(f消费: {item}) time.sleep(0.3) threads [ threading.Thread(targetproducer), threading.Thread(targetconsumer), ] for t in threads: t.start() q.join() # 等队列中所有任务被处理完 # 生产结束后放哨兵注意q.join()和task_done()的配合使用。Queue内部维护一个未完成任务计数每次get()后调用task_done()才把计数减一join()才会返回。很多新手漏了task_done()导致join()永远阻塞。4.3 死锁的典型场景与排查思路死锁的经典场景是互相持有对方需要的锁。A 线程持有锁 1 等锁 2B 线程持有锁 2 等锁 1两边互不相让程序永远卡死。import threading import time lock_a threading.Lock() lock_b threading.Lock() def task_a(): with lock_a: time.sleep(0.1) # 故意让两个线程撞上 with lock_b: print(task_a done) def task_b(): with lock_b: time.sleep(0.1) with lock_a: print(task_b done) t1 threading.Thread(targettask_a) t2 threading.Thread(targettask_b) t1.start() t2.start() t1.join() t2.join() # 程序卡死在这里这个代码跑起来就是教科书级的死锁。排查手段我推荐三个层次faulthandler.dump_traceback_later()在死锁场景下定时 dump 所有线程的堆栈。这是标准库内置的不需要装额外工具我最常用。py-spy dump --pid PID第三方工具可以附着到运行中的进程打印所有线程当前的堆栈对排查线上死锁特别有效。缺点是 Linux 下需要一定权限。pdb手动断点适合本地复现断点打在可疑的acquire附近手工看线程状态。写代码时预防死锁的几条经验锁的获取顺序保持一致。多个锁嵌套时全局统一顺序比如都先拿 A 再拿 B。尽量缩小临界区锁内不做耗时操作尤其不要做 I/O网络请求、文件写。用acquire(timeout...)给锁加超时获取不到就放弃或者重试至少不会无限卡死。4.4 线程本地数据不想共享就别共享有些数据每个线程应该各有一份不需要也不应该共享比如每个线程独立的数据库连接、会话、随机数种子。这时候用threading.local()再合适不过。import threading local_data threading.local() def worker(name): local_data.thread_name name # 每个线程独立设置互不干扰 print(f{threading.current_thread().name}: {local_data.thread_name}) threading.Thread(targetworker, args(线程A,)).start() threading.Thread(targetworker, args(线程B,)).start()local_data在不同线程里访问到的值是隔离的类似一个线程维度的字典。在很多并发框架包括 web 框架里用它来保存请求级别的上下文比全局变量安全得多。5. 几个容易踩的生产级细节纸上谈兵差不多了聊几个我在真实项目里踩过的坑。这些坑不大但每一个都足以让你排查大半天。5.1 守护线程daemon的退出陷阱threading.Thread(daemonTrue)表示这是一个守护线程。主线程退出时守护线程会被强制终止不执行清理、不触发finally、不做任何善后。区分需求后台任务如果允许主程序退出就放弃可以设daemonTrue但如果这个线程负责写日志、刷新缓存、优雅关闭连接那绝对不能设守护必须join()等它干完。我最常犯的错是写个轮询线程想着反正是后台任务就设了daemonTrue结果程序退出时还会收到 SIGTERM导致日志文件里的最后几条记录丢了。改成非守护线程后主流程join()等待这个线程正常退出问题就解决了。5.2 异常在子线程里是沉默的子线程抛出异常主线程完全无感知——程序不会报错主线程该干嘛干嘛。这个坑特别隐蔽因为本地测试时线程启动的异常可能还能在控制台看到一旦上了服务框架子线程异常直接进日志黑洞。我的习惯是在线程入口包一层try/except手动把异常传给主线程或写到专门的错误队列import threading import queue error_q queue.Queue() def safe_worker(): try: # 实际业务逻辑 raise ValueError(演示异常) except Exception as e: error_q.put(e) t threading.Thread(targetsafe_worker) t.start() t.join() if not error_q.empty(): err error_q.get() print(f子线程捕获到异常: {err})如果有第三方代码在线程里执行一旦不确定是否静默吞异常最好统一走这个模式。5.3 线程创建的爆炸式增长防不胜防每来一个请求就开一个线程这种代码我在很多项目的早期版本里都见过。请求量一大线程数飙升到几千甚至上万每个线程还要占 8MB 虚拟栈空间、消耗上下文切换资源服务直接假死。上文提到的ThreadPoolExecutor是标准解但还有一个细节要注意如果任务队列里积压了大量任务线程池再大也是杯水车薪。此时应该做的是限流和削峰比如用信号量限制同时处理的任务数或者直接上消息队列做缓冲。线程池解决的是并发执行问题不解决任务洪水问题这两个概念千万别混。5.4 文件写入乱序别让多个线程抢同一个句柄多线程同时往同一个文件里写日志写出来的内容会乱成一团——不是内容丢了而是交错穿插。因为文件对象本身不是线程安全的。解决方案有三类所有写文件操作集中到一个线程其他线程把日志内容丢进queue.Queue由这个线程统一写。用日志库logging.handlers里内置的锁机制logging模块的FileHandler本身是线程安全的。真需要高性能并发写考虑每个线程独立文件异步合并。我强烈推荐第一种或者第二种。手动加锁写文件也能跑但会在写文件这种低耗时操作上引入锁竞争吞吐量在日志量大时下降明显。6. 线程与协程、多进程怎么配合使用线程不是 Python 并发世界的唯一答案。协程asyncio、多进程multiprocessing各有各的适用场景但实际项目里它们经常不是互斥关系而是协同关系。6.1 选型逻辑看你的瓶颈在哪瓶颈推荐方案核心原因网络 I/O、文件 I/O 等待协程或线程等待期让出执行权复用率高CPU 密集计算多进程绕开 GIL真正利用多核两者混合多进程 线程/协程各司其职计算和等待分离线程和协程的选择上我的经验是如果项目从零开始、主要就是高并发 I/O 场景而且团队熟悉异步编程asyncio的吞吐量上限更高因为它没有线程的上下文切换和内核态开销。但如果业务逻辑里有大量同步的第三方库调用比如很多数据库驱动、老版本 requests硬上asyncio反而痛苦线程是更务实的选择。这里有个实用技巧在asyncio事件循环里调用同步阻塞库时用loop.run_in_executor(None, func, args)把它丢到线程池里执行这样既保持了异步架构又不会阻塞事件循环。这是我处理异步框架 同步库兼容问题的最常用招数。import asyncio import time def blocking_query(x): time.sleep(1) return x * 2 async def main(): loop asyncio.get_running_loop() # 把阻塞操作丢到线程池避免阻塞事件循环 result await loop.run_in_executor(None, blocking_query, 21) print(result) asyncio.run(main())6.2 线程 进程的混合模式我做过一个写爬虫的项目目标站点多、单次请求耗时高、返回数据还需要做大量 JSON 解析和清洗。如果纯用线程解析阶段抢 GIL 拖慢整体如果纯用进程进程之间的数据传递成本高。最后是混合模式multiprocessing开 4 个进程每个进程内部用ThreadPoolExecutor跑 10 个线程做请求和初步解析解析结果通过multiprocessing.Queue汇总到主进程做最终处理。这样请求等待和 CPU 解析都得到了最大的资源利用。这个方案实现复杂度更高但吞吐量比纯线程版本高了将近一倍。如果你的任务同时具备I/O 等待 CPU 计算两种特征可以考虑这个方向。7. 实测总结线程并发的数量红线与性能参考最后给点硬核参考。我粗略统计了近期多个项目在真实负载下的线程模型数据谈不上严谨压测但给你一个不太会跑偏的感觉机器配置线程场景实测参考4 核 8G 云主机线程池 I/O 密集任务max_workers20~50工作良好16 核 32G 物理机线程池 I/O 密集任务max_workers100~200可达吞吐顶峰8 核桌面 PCCPU 密集任务 线程线程数超过 2 就开始明显拖慢任何配置手动每请求一线程线程数 500 后上下文切换开销显著延迟飙升注意这些数字会因具体任务特征浮动不能当作万能公式抄。我的建议是I/O 密集场景从CPU 核数 × 10起步跑一轮真实流量看 CPU 延迟曲线再逐步调高千万别一上来拍脑袋定个大数。7.1 线程数的调优方法一个笨但可靠的方法是做个简单的阶梯压力测试固定任务量比如 1000 个模拟请求分别设置max_workers10/20/50/100/200。记录总耗时和每个请求的平均延迟。画个趋势总耗时先降后升拐点就是你的最优并发数。以我的经验I/O 密集任务的最优并发数通常远大于 CPU 核数因为线程都在等网络包但超过一定阈值后线程切换和 GIL 争抢成本会盖过并发收益。那个天花板在哪只有实测才能找到。7.2 线程与 I/O 事件循环的对比感受用同样的任务量分别用线程池和asyncio跑我的经验数据是纯 I/O 场景下asyncio的开销大概是线程的 1/51/3内存占用和 CPU 占用都更省。但代码复杂度会明显上升尤其是在处理限流、重试、超时这类业务逻辑时。我的建议很实际团队熟哪个用哪个别为了高性能盲目换异步框架。并发编程的第一目标是正确和可维护性能提升建立在正确的架构之上。用线程把并发跑通、保证正确性再考虑要不要换成异步来优化这个迭代路径远比一上来就上asyncio稳。8. 经验之谈线程调试的几把土武器说了这么多理论和数据最后留点私货——我在排查线程问题时的几个土办法。工具未必高大上但确实帮我解决过不少头疼的问题。8.1 日志里带线程名和线程 ID多线程环境打日志不带线程信息等于白打。我固定用threading.current_thread().name和threading.get_native_id()拼进日志前缀这样一出问题翻日志就能按线程维度筛。import threading import logging logging.basicConfig( levellogging.INFO, format%(asctime)s [%(threadName)s] [tid%(threadid)s] %(message)s, )threadName可以直接用tid需要额外一个LogRecord过滤器传入或者直接在日志消息里手动拼。日志配合线程维度排查并发问题至少能筛掉一半的障碍。8.2 用 faulthandler 抓卡死现场faulthandler是标准库但知道的人不算多。它可以在程序卡死的状态下定期把每个线程的 Python 堆栈打印到 stderr 或者文件里。import faulthandler import threading import time faulthandler.dump_traceback_later(5, repeatTrue) # 5 秒后开始 dump持续重复 def deadlock_task(): lock threading.Lock() lock.acquire() lock.acquire() # 自我死锁 threading.Thread(targetdeadlock_task).start() time.sleep(30)运行时你能看到每个线程当前停在哪个函数哪一行立刻定位卡点在哪儿。如果你在云服务器上排查问题这招比 GDB 好使太多不需要额外装任何东西。8.3 用py-spy看线上进程py-spy是我最常用的第三方调试工具比gdb方便不需要程序以调试模式运行。py-spy dump --pid PID直接打印那个进程里所有线程的 Python 堆栈。遇到线上卡死attach 上去瞬间就知道线程卡在锁、网、IO 的哪一层。注意py-spy在容器里运行时需要一定的权限配置特定 Linux 发行版可能还要设置ptrace_scope。但为了线上排查问题值得提前配好。8.4 复现竞态条件的小技巧竞态条件往往不稳定复现今天跑得好好的明天就炸。我的技巧是在关键临界区前后故意加一点随机阻塞把竞态的概率窗口人为拉大让 bug 更容易暴露。import random import time def flaky_worker(): time.sleep(random.uniform(0.001, 0.01)) # 进入临界区...这个技巧没法用来修复 bug但用来验证是不是并发问题一测一个准。真正修复还是得靠正确的同步原语和设计。最后再分享一个小技巧写并发代码永远先从最朴素的正确方案开始比如加一把大锁保证逻辑正确然后再打磨性能比如缩小临界区、换读写锁、改无锁结构。我见过太多人一上来就追求无锁设计结果花了大量时间调 bug最后还不如一把锁来得快。并发编程里正确性先于性能这句话什么时候都不过时。