直接聊结论Python多线程到底能不能提升性能答案取决于你干什么。这是每个Python开发者在入门之后都会撞上的一堵墙。很多人写爬虫发现多线程飞快写计算任务发现线程开着比单线程还慢然后又听说了GIL这个东西反而更迷茫了。这篇文章就围绕Python多线程的核心机制、常用工具、实践场景和踩坑经验把来龙去脉讲清楚。如果你现在正打算用threading、ThreadPoolExecutor处理一批任务或者正在纠结该用线程还是协程这篇内容值得你从头看一遍。我会尽量用最直白的话拆解GIL这些底层概念再配上可以直接抄走的代码套路。1. 多线程的底层逻辑先理解为什么Python线程“特别”1.1 GIL到底是什么GIL全称Global Interpreter Lock全局解释器锁。这是CPython解释器的一个核心机制。你在Windows上装的python.org版本Linux上大多数发行版自带的版本都是CPython。所以只要用官方Python这个问题躲不掉。GIL的规则非常简单粗暴在任意一个时刻同一个进程里只有一个线程在真正执行Python字节码。换句话说Python的多线程不是真正意义的多核并行而是多线程轮流抢占一个执行名额。很多人第一次听到这里会问既然这样多线程还有什么用答案是有用而且非常有用关键看你的任务属于哪种类型。GIL之所以存在根源在于CPython的内存管理机制。Python用引用计数来管理对象生命周期当一个对象的引用计数变成0就立即回收内存。如果允许多个线程同时操作引用计数计数可能被同时修改内存安全问题随之而来。最极端的解法就是在解释器层面加一把全局锁牺牲掉并行性换取内存管理的绝对安全。后来PyPy曾尝试做STM版本的Python去掉GIL但没有在主流场景铺开。Python 3.13引入了free-threaded模式去掉了GIL但那是很新的特性目前生产环境的普及程度还很低。所以现阶段你写业务代码还是需要把GIL当作一个客观存在的约束。1.2 GIL对I/O密集和CPU密集任务的影响I/O密集任务指那些大部分时间在等待外部资源的任务——等网络响应、等文件读写、等数据库查询结果。这些操作的共同特点是线程进入等待状态时并不会持有一个叫做“正在执行字节码”的状态锁可以释放给其他线程用。简单说爬虫就是典型例子。一个线程发HTTP请求发出之后等着服务器返回。这段等待时间里CPU几乎是空闲的。如果单线程做就只能干等着开多线程一个线程等网络的时候另一个线程可以继续发请求。整体看多线程在I/O密集场景下的加速效果非常明显。CPU密集任务则相反。比如计算矩阵乘法、跑图像滤波算法、循环做大量数值运算这些操作全程都需要CPU不断执行。这时GIL的存在意味着多个线程根本没法同时推进计算反而是线程切换带来了额外开销。结果就是线程开得越多总耗时越长。提示判断该不该用多线程先看你的任务是吃CPU还是吃I/O。吃I/O的用线程吃CPU的想追求并行就要考虑多进程或切换计算库来实现。2. 线程创建与生命周期管理2.1 最基础的threading模块用法Python标准库的threading模块提供了创建和管理线程的全部基础设施。最常用的就是Thread类。一段最基础的多线程代码长这样import threading import time def worker(name: str, delay: float): print(f线程{name}启动) time.sleep(delay) print(f线程{name}结束) threads [] for i in range(5): t threading.Thread(targetworker, args(fworker-{i}, 1)) threads.append(t) t.start() for t in threads: t.join() print(所有线程执行完毕)这里有三个关键操作start()让线程进入可运行状态join()让主线程阻塞等待子线程结束args传递参数到目标函数。看起来很简单但实际工程中光靠这个还不够。需要注意start()并没有让线程立刻执行只是告诉解释器这个线程可以开始参与了。具体什么时候真正跑到你给定的target函数由解释器线程调度决定。所以不要期望start()之后立刻能读到线程里的变量值。2.2 守护线程与线程状态管理threading.Thread有个daemon属性这个很多人会忽略但非常重要。t threading.Thread(targetworker, args(demo, 1), daemonTrue) t.start()daemonTrue表示这个线程是守护线程。守护线程的特点是当所有非守护线程都结束后守护线程会被强制终止整个进程退出。也就是说进程不等待守护线程执行完。看下面这个例子import threading import time def long_task(): time.sleep(10) print(任务完成) t threading.Thread(targetlong_task, daemonTrue) t.start() print(主线程结束)运行以上代码你会发现主线程打印完“主线程结束”之后进程立刻退出10秒后的那条print根本没机会执行。因为主线程是非守护线程它结束了守护线程强制结束程序退出。如果不用daemonTrue默认非守护线程会让进程一直等到10秒后才退出。实践中的经验是对于周期性的后台任务比如定时刷新缓存、心跳检测用守护线程合适对于真正需要完整执行才能保证状态一致性的任务不要设守护线程。线程还涉及状态管理。线程有新建、就绪、运行、阻塞、终止几个状态。Python层面你只能间接感知is_alive()判断线程是否还活着enumerate()列出当前所有线程active_count()获取线程总数但需要注意is_alive()返回True只能说明线程对象还活着不代表你的任务已经完成了一半。你更需要的其实是事件等待和队列消费进度。2.3 线程安全的切入点以及何时需要加锁多个线程同时读写同一个对象如果没有保护就可能出现数据错乱。典型例子多个线程同时向同一个列表追加数据或者多个线程同时修改同一个计数器。看个经典案例import threading counter 0 def increment(): global counter for _ in range(1000000): counter 1 threads [threading.Thread(targetincrement) for _ in range(10)] for t in threads: t.start() for t in threads: t.join() print(counter) # 大概率不是10000000counter 1看起来只有一步但字节码层面实际是“读取→加一→写回”三个动作。线程A读到了当前值5还没写回6的时候线程B也读到了5两边同时加一最终写回时覆盖了彼此的结果。这个现象叫竞态条件。要解决给读写操作加锁import threading counter 0 lock threading.Lock() def increment(): global counter for _ in range(1000000): with lock: counter 1with lock保证同一时刻只有一个线程能够执行counter 1其他线程必须等待。最终counter的结果稳定为10000000。我的建议是不要为了避免加锁而绕过问题也不要每行代码都加锁。加锁的粒度要小而精准只锁住需要保护的那几行代码。锁范围越大线程之间互相等待的时间越长性能越差。3. 锁与同步多线程协作的正确姿势3.1 Lock与RLock的区别和选型threading.Lock是一把标准互斥锁。它有一个规则锁不能被同一个线程重复获取。你在同线程里连续lock两次第二次会把自己阻塞死。这就是死锁。import threading lock threading.Lock() with lock: # 同一个线程再次获取就会卡死 with lock: pass上面的代码会直接卡住因为同一个线程在持锁没释放时又尝试获取锁。如果代码天然存在嵌套锁的场景用RLock可重入锁import threading lock threading.RLock() with lock: with lock: passRLock允许同一个线程多次获取锁内部维护了一个计数器和线程持有者信息。只有获取锁的线程才能释放且必须等所有acquire都被release之后才算真正释放。写递归调用中需要加锁的代码时RLock非常方便。实际选型标准很简单你的代码里有没有可能在同一线程内重复获取同一把锁。有可能就用RLock。只是保护一段简单代码不涉及嵌套就用Lock性能略好一点点。3.2 Condition、信号量与事件的适用场景Lock解决了互斥问题但线程之间还经常需要“等待某个条件满足后继续”。比如生产者生成了商品消费者才能拿走。轮询判断也可以但太低效。标准库提供了Condition、Event、Semaphore等同步工具。Event适合“一个线程通知多个线程”的场景import threading import time event threading.Event() def waiter(name): print(f{name}等待信号...) event.wait() print(f{name}收到信号继续执行) def setter(): time.sleep(2) print(发出信号) event.set() threading.Thread(targetwaiter, args(A,)).start() threading.Thread(targetwaiter, args(B,)).start() threading.Thread(targetsetter).start()event.wait()会阻塞线程直到set()被调用。这个模式很适合做“所有子线程都准备好了主线程再放行”之类的流程控制。Condition更适合生产者-消费者模型。它允许线程等待某个条件同时条件不满足时自动释放锁import threading import time buffer [] condition threading.Condition() MAX_SIZE 5 def producer(): for i in range(10): with condition: while len(buffer) MAX_SIZE: condition.wait() buffer.append(i) print(f生产了 {i}) condition.notify_all() time.sleep(0.1) def consumer(): for _ in range(10): with condition: while not buffer: condition.wait() item buffer.pop(0) print(f消费了 {item}) condition.notify_all() time.sleep(0.2) threading.Thread(targetproducer).start() threading.Thread(targetconsumer).start()这里的关键点是condition.wait()会暂时释放锁让其他线程有机会进入修改状态。等到notify被调用等待线程被唤醒后重新获取锁继续执行。Semaphore则用来限制并发数量。比如同时最多3个线程访问某个资源超出就阻塞semaphore threading.Semaphore(3)这个做限流、资源池控制时非常实用。4. 队列与线程池把线程用得优雅且高效4.1 queue.Queue如何简化线程协作自己管理线程之间的数据传递总是要加锁容易出现疏忽。queue.Queue本身就在内部帮你实现了锁和同步逻辑直接用就可以。典型的线程池协作模型import queue import threading import time import random task_queue queue.Queue() def worker(name): while True: task task_queue.get() if task is None: break print(f[{name}] 处理任务 {task}) time.sleep(random.random()) task_queue.task_done() def boss(): for i in range(10): task_queue.put(i) task_queue.join() worker_count 3 threads [] for i in range(worker_count): t threading.Thread(targetworker, args(fworker-{i},)) t.daemon True t.start() threads.append(t) boss() for _ in range(worker_count): task_queue.put(None) for t in threads: t.join()这里有两个要点。第一使用task_done()和join()配合join会阻塞到所有任务都标记为完成。第二用哨兵值None通知工作线程退出避免了线程永远卡在get()上。queue.Queue默认无界生产速度远大于消费速度时内存会不断膨胀。指定maxsize限制队列大小task_queue queue.Queue(maxsize100)队列满时put()会阻塞直到有空位。这样自然做了背压控制。4.2 ThreadPoolExecutor实战Python 3.2之后concurrent.futures模块提供了ThreadPoolExecutor比手动管理threading线程省心很多。你不需要负责start、join、任务队列只需要提交任务等结果。基础用法from concurrent.futures import ThreadPoolExecutor, as_completed import time def fetch_data(url): time.sleep(1) return fdata from {url} urls [fhttps://api.example.com/{i} for i in range(20)] with ThreadPoolExecutor(max_workers5) as executor: future_to_url {executor.submit(fetch_data, url): url for url in urls} for future in as_completed(future_to_url): data future.result() print(data)这段代码会以5个并发线程处理20个任务。with语句块结束时会等待所有任务完成并自动关闭线程池。map方法更简洁with ThreadPoolExecutor(max_workers5) as executor: results list(executor.map(fetch_data, urls))executor.map会按提交顺序返回结果适合需要保持顺序的场景。但它的缺点是如果某个任务卡住后面的结果也不会返回除非设置超时for future in executor.map(fetch_data, urls, timeout5): print(future)submit配合as_completed与直接map相比as_completed是有结果就返回map是严格按顺序。根据实际场景二选一就行。捕获异常也很重要。future.result()的任务在后台抛出异常时不会直接打印而是等你调用result()时才抛出try: result future.result(timeout3) except Exception as exc: print(f任务异常: {exc})这样能清晰定位到具体是哪个任务出问题。4.3 池化背后的资源管理逻辑为什么要用线程池而不是每来一个任务就创建一个新线程线程的创建和销毁有成本。每创建一个线程系统要分配栈内存、初始化线程控制块、注册调度信息。在Linux上默认线程栈一般是8MB的虚拟内存空间如果任务量很大大量短生命周期线程会浪费大量系统资源。线程池的核心思想是复用固定数量的线程。任务来了线程从池里被调度去处理任务做完线程继续等待新任务而不是被销毁。这样省掉了反复创建销毁的开销。线程池大小的确定没有绝对公式。经验法则是I/O密集任务线程数可以略多通常是CPU核心数的5到10倍CPU密集任务线程数建议等于CPU核心数混合任务看哪种占比更高再调整Python的os.cpu_count()可以获取逻辑CPU核心数import os cpu_count os.cpu_count() print(cpu_count)不过说实话实际项目中最好的方式还是做一次小规模压测把不同max_workers跑出来的耗时对比一下再定。注意线程池虽然好用但如果你的任务里包含CPU计算密集的部分多线程加速有限这时候就该考虑ProcessPoolExecutor或直接换计算方案。5. 协程与多线程的边界5.1 协程vs线程什么是真正的替代方案很多人纠结该学协程还是学线程或者在项目里到底用哪个。其实二者解决的问题有重叠但机制完全不同。线程是操作系统层面的调度单位由内核负责切换。协程是用户态调度的机制切换不依赖系统调用开销更小。用协程实现并发核心是需要配合异步I/O。比如用asyncio写网络请求一个事件循环在等待请求返回的空闲期会自动切换去执行其他协程任务这种模式下不需要线程也不需要GIL介入。做一个对比你就看得很清楚# 多线程版本 import threading import time def download(url): time.sleep(1) return url urls [fpage-{i} for i in range(10)] start time.time() with ThreadPoolExecutor(max_workers10) as ex: list(ex.map(download, urls)) print(f多线程耗时: {time.time() - start:.2f}秒)# 协程版本 import asyncio async def download_async(url): await asyncio.sleep(1) return url async def main(): urls [fpage-{i} for i in range(10)] await asyncio.gather(*(download_async(url) for url in urls)) start time.time() asyncio.run(main()) print(f协程耗时: {time.time() - start:.2f}秒)两个版本耗时都接近1秒。区别在于多线程版本中每个线程都要占用操作系统资源协程版本只是在一个线程内来回切换。并发越多协程的优势越明显。5.2 爬虫场景的最优实践写爬虫是最典型的I/O密集场景。我在实际项目中总结了一条经验数据量小、并发要求低用线程池最简单数据量大、需要超高并发用协程更合适。协程写法对第三方库有要求。requests是同步库直接放在async函数里会阻塞事件循环反而起不到并发效果。需要用aiohttp或者httpx的async接口。import asyncio import aiohttp async def fetch(session, url): async with session.get(url) as resp: return await resp.text() async def main(): async with aiohttp.ClientSession() as session: urls [fhttps://api.example.com/item/{i} for i in range(50)] tasks [fetch(session, url) for url in urls] results await asyncio.gather(*tasks) print(f完成 {len(results)} 个请求)如果你不想引入太多异步依赖线程池配合requests也完全够用。限制并发数的话ThreadPoolExecutor的max_workers本身就是限流机制。协程里限流则需要用semaphoresemaphore asyncio.Semaphore(10) async def fetch_with_limit(session, url): async with semaphore: return await fetch(session, url)关于什么时候用线程、什么时候用协程记住一句人话你的代码是在等外部资源网络、磁盘、数据库线程够用等的数量极大且要求高吞吐上协程。永远不要为了炫技强行切换技术方案。6. 实战中的线程数选择与疑难排查6.1 如何确定线程池的最佳大小网上流传很多公式但我实际测下来发现理论值只能当起点。受限于GIL、服务端限流、网络带宽、目标服务的响应能力线程池大小需要针对具体业务调整。一个适合用于估算的经验公式是I/O密集最佳线程数 CPU核心数 × (1 平均等待时间 / 平均计算时间)CPU密集最佳线程数 CPU核心数 1举个例子你的任务平均在CPU上计算0.2秒等待外部接口1.8秒那么8核机器上最佳线程数大约是8 × (1 1.8 / 0.2) 80。这个结果和“线程数是核心数5到10倍”的经验一致。实际操作中可以先设置几个档位做压测比如max_workers分别取10、20、40、60对比总耗时和错误率。这样比只看理论公式可靠得多。遇到目标服务有限流的情况线程数设置太大反而会触发大量429响应或连接超时。这时线程池大小反而要控制得比理论值小配合重试策略。6.2 常见死锁、崩溃与资源耗尽问题排查多线程写代码容易写对难。最常见的坑我整理了下面几类。第一死锁。死锁的典型特征是程序在某个点完全卡住没有任何报错。排查死锁可以按顺序按下CtrlZWindows下或使用pdb检查当前线程状态faulthandler.enable()可以在卡住时打印所有线程的堆栈检查是否存在两个锁互相等待线程A持有锁1等待锁2线程B持有锁2等待锁1import faulthandler import threading import time faulthandler.enable() lock1 threading.Lock() lock2 threading.Lock() def task_a(): with lock1: time.sleep(0.1) with lock2: pass def task_b(): with lock2: time.sleep(0.1) with lock1: pass运行后卡住时按CtrlC就能看到当前两个线程分别持有哪把锁、在等哪把锁。第二线程太多导致的资源耗尽。创建大量线程时可能出现异常cant start new thread。这是因为进程内线程数达到系统限制或者进程地址空间不足。排查手段用ulimit -u查看用户最大进程数Linux检查是否有线程没被正确join出现了泄漏pstree -p或者ps -eLf查看线程数量第三内存爆炸。queue.Queue没设置maxsize生产速度远大于消费速度队列无限堆积内存持续上涨。典型的解决方式是设置maxsize并利用put(blockTrue)的阻塞特性做背压。第四虚假并发导致的性能劣化。一个本身设计成连续I/O的函数如果内部已经用了异步方式外面包装成多线程不会带来收益。还有某些重复代码里频繁acquire/release锁锁竞争本身就会成为瓶颈。我用py-spy dump采集过线上进程发现大量线程阻塞在acquire锁等待上那基本就说明锁粒度太大了。排查这种隐性性能问题时py-spy是一个非常实用的工具pip install py-spy py-spy dump --pid 进程ID这个命令能把进程内所有线程的当前堆栈都打印出来不用重启进程。相比手动猜到底卡在哪一行效率高得多。线程安全问题还有一个容易被忽视的点logging模块。Python的logging是线程安全的但很多第三方库内部写了不可重入的全局状态。跨线程共享一个session对象或者在多线程里使用同一个游标都是高危操作。我的处理原则是一个线程一个连接/一个会话不共享。代码里还有一个细微但容易踩的坑items [] def process(): local_items items.copy() # 只操作local_items即使每个线程只操作局部变量如果局部变量引用了全局对象的内部结构仍然可能产生并发写。要区分清楚“线程局部”和“对象局部”。实践中我尤其建议给所有多线程代码加超时控制。future.result(timeout5)、queue.get(timeout5)这类东西看起来繁碎但生产环境里一次外部服务无响应就能拖垮整条线程池。最后分享一个小经验多线程程序里能放在外面初始化的资源尽量在主线程提前初始化。有的第三方库的初始化逻辑里有静态内部状态并发初始化时会莫名其妙崩溃。线程内只做任务不做环境准备稳定性会高一个量级。
)
