如果你刚开始学 Python 并发threading模块往往是最先接触的东西。它解决的不是“让代码看起来更酷”而是让程序在等待网络请求、文件下载、接口返回这类耗时操作时不要一直傻等而是尽量去处理别的任务。很多初学者第一次写多线程都会遇到几个典型现象线程启动顺序很乱、主线程提前跑完了、共享变量结果不对、程序结束后迟迟不退。第 37 天的内容我就围绕threading模块把这些问题全部拆开讲一遍重点放在能落地运行的基础代码上。先说一个最关键的判断threading真正适合的是 IO 密集型任务不太适合用来加速纯 CPU 计算。Python 解释器里存在 GIL纯计算场景下多线程很难吃满多核 CPU。如果你正计划用 Python 写爬虫批量下载、批量调用接口、处理大量小文件那么threading的思路值得学如果想做图像算法、数学计算之类的并行加速应该去看multiprocessing或其它方案。1. 多线程适合什么任务先分清 IO 密集和 CPU 密集1.1 网络请求和文件下载为什么适合多线程IO 密集型任务的特点是大量时间花在“等待”上。比如一个网络请求发出后客户端真正需要做的工作是等待服务端响应一个文件读取任务发出后读取操作会等待磁盘数据返回。这个等待过程里CPU 本身没有太多事情要做。如果不使用多线程程序只能按顺序执行import time def request_api(name): print(f{name} 开始请求) time.sleep(2) # 模拟网络请求耗时 print(f{name} 请求完成) start time.perf_counter() request_api(任务A) request_api(任务B) print(总耗时, time.perf_counter() - start)上面这段代码的总耗时约为 4 秒因为任务 B 必须等到任务 A 结束后才能开始。改成多线程后任务 A 在等待期间任务 B 也有机会被调度执行。同样是 2 个任务理想情况下总耗时接近 2 秒而不是 4 秒。爬虫批量下载、批量查询数据库、批量调用第三方接口都是这种模式。这也是为什么“Python 多线程”经常和爬虫、文件下载、接口请求联系在一起。1.2 Python 的 GIL 怎么影响多线程效果GIL 全称 Global Interpreter Lock是 CPython 解释器里的一个全局锁。它保证同一时刻只有一个线程能执行 Python 字节码。也就是说在多核 CPU 上多个 Python 线程并不能真正同时跑 Python 代码。那为什么多线程还能提升 IO 任务速度因为线程在等待网络、等待磁盘、调用某些底层 C 库的时候会释放 GIL。等待期间不需要 Python 解释器继续干活其它线程就可以拿到 GIL 继续执行。所以多线程提升的是“总体吞吐时间”不是把每一行 Python 代码的执行速度翻倍。1.3 纯 CPU 计算为什么不适合用 threading如果你有一大段纯 Python 循环计算想用多线程加速def calculate(): result 0 for i in range(10000000): result i return result这种情况下多线程不会带来明显提升。因为每个线程都要抢同一个 GIL切换线程反而会增加额外开销。正确的做法是用multiprocessing创建多个进程每个进程都有独立的 Python 解释器和内存空间才能真正利用多核 CPU。如果任务是高并发 IO比如成千上万个网络连接也可以考虑asyncio。在学习threading之前先想清楚这一点能避免后面出现“明明加了线程速度却没有变快”的困惑。2. threading 模块入门从线程对象到生命周期控制2.1 最小可运行示例创建线程并启动threading模块最基础的用法就是创建Thread对象。它接收一个target参数指向线程要运行的函数。import threading import time def worker(name): print(f线程 {name} 开始) time.sleep(1) print(f线程 {name} 结束) if __name__ __main__: t1 threading.Thread(targetworker, args(A,)) t2 threading.Thread(targetworker, args(B,)) t1.start() t2.start() t1.join() t2.join() print(主线程执行完毕)执行这个程序正常会看到线程 A 和 B 都有输出并且最后打印“主线程执行完毕”。args用于给目标函数传参必须是元组。如果只有一个参数要写成args(A,)不能漏掉逗号。2.2 start、join、daemon 分别控制什么start()的作用是让线程进入可调度状态。它不是立刻执行目标函数而是向操作系统注册线程等待调度。所以代码里先调用start()的线程不一定最早打印。join()的作用是让当前线程等待指定线程结束。主线程调用t1.join()就会等线程 t1 执行完再继续往下走。如果不调用 join主线程可能会先打印“执行完毕”但子线程仍然在运行。还有一个需要注意的地方不要在循环里一边start()一边立刻join()。# 这种写法接近串行 for i in range(5): t threading.Thread(targetworker, args(i,)) t.start() t.join()因为每次启动一个线程后立刻等它结束后面的线程还没启动前面线程已经跑完了并发效果会被削弱。正确做法是先把所有线程对象保存到列表里然后统一join()threads [] for i in range(5): t threading.Thread(targetworker, args(i,)) threads.append(t) t.start() for t in threads: t.join()daemon参数也很容易踩坑。默认情况下它是False意思是主线程结束后程序会等待所有非守护线程执行完再退出。如果设置daemonTrue主线程结束时不会等待该线程程序可能直接退出。t threading.Thread(targetworker, args(C,), daemonTrue) t.start()如果线程里有长任务并且你没让它join()daemon 线程可能没跑完程序就结束了。守护线程适合后台心跳、日志上报这类可随时中断的任务不适合关键数据处理。3. 代码实战把一个批量下载任务改成多线程3.1 先写单线程版本这里模拟一个批量下载场景有一个 URL 列表每个 URL 都对应一个文件。为了便于演示用time.sleep()模拟下载耗时。import time import random files [fhttps://example.com/file/{i}.zip for i in range(6)] def download_one(url, index): cost random.uniform(0.2, 0.8) time.sleep(cost) print(f任务 {index} 下载完成耗时 {cost:.2f}s) start time.perf_counter() for index, url in enumerate(files): download_one(url, index) print(f单线程总耗时{time.perf_counter() - start:.2f}s)这个版本逻辑很清晰下载完第一个文件再下载第二个。问题是如果每个文件下载需要 1 秒6 个文件就要大约 6 秒。3.2 多线程改造先创建线程再统一等待改造逻辑不复杂把每个下载任务放进一个线程对象里启动所有线程最后统一join()。import threading import time import random files [fhttps://example.com/file/{i}.zip for i in range(6)] def download_one(url, index): cost random.uniform(0.2, 0.8) time.sleep(cost) print(f任务 {index} 下载完成耗时 {cost:.2f}s) threads [] start time.perf_counter() for index, url in enumerate(files): t threading.Thread(targetdownload_one, args(url, index)) threads.append(t) t.start() for t in threads: t.join() print(f多线程总耗时{time.perf_counter() - start:.2f}s)这里最核心的步骤是先启动所有线程再循环join()。这样多个下载任务才能并发执行。如果你在真实项目里是下载文件到本地记得每个线程的输出文件名不要重复建议在文件名里带上任务编号比如file_0.zip、file_1.zip。否则两个线程同时写同一个文件内容容易被覆盖而且很难排查。3.3 怎么判断多线程有没有真的生效最直接的判断标准是总耗时。单线程总耗时约等于所有任务耗时之和多线程总耗时接近最慢任务耗时加上少量线程切换开销。还有一个细节是打印顺序。多线程版本里输出顺序可能不是从 0 到 5 的。比如先看到任务 3 完成再看到任务 1 完成。这是正常的因为每个任务耗时不同线程完成时间不同。如果多线程版本总耗时反而比单线程慢优先检查三件事任务是不是 CPU 密集而不是 IO 密集线程之间是否为了争抢锁或资源大量等待是不是在循环里启动一个就立刻 join 一个把并发退化成了串行。4. 多线程访问共享数据Lock 是必修课4.1 一次计数竞争实验结果不等于预期多线程最麻烦的问题不是线程创建而是多个线程同时访问同一个变量。看下面这段代码import threading counter 0 def worker(): global counter for _ in range(200000): counter 1 threads [] for _ in range(4): t threading.Thread(targetworker) threads.append(t) t.start() for t in threads: t.join() print(counter 结果, counter)理想情况下4 个线程各加 200000 次结果应该是 800000。但实际上结果经常小于这个数而且每次运行可能都不一样。原因是counter 1并不是一步完成的。它包含了读取当前值、计算加一、写回新值三个步骤。线程 A 读取到 100线程 B 也读取到 100A 写回 101B 也写回 101结果就少了一次更新。这种问题叫竞态条件也叫资源竞争。4.2 用 Lock 保护临界区解决方式是在修改共享变量之前加锁。同一时刻只有一个线程能进入加锁范围其它线程必须等待锁释放。import threading counter 0 lock threading.Lock() def worker(): global counter for _ in range(200000): with lock: counter 1 threads [] for _ in range(4): t threading.Thread(targetworker) threads.append(t) t.start() for t in threads: t.join() print(counter 结果, counter)with lock等同于先调用lock.acquire()执行完代码块后自动调用lock.release()。它比手动 acquire 和 release 更安全避免因为异常导致锁没有释放。加锁后的结果会稳定等于 800000代价是执行速度会下降。因为每个线程修改共享变量时都需要排队。所以写多线程程序时要有一个取舍不是所有代码都要加锁而是只给共享资源相关的那一小段代码加锁。锁的范围太大会让多线程退化回串行。4.3 Lock 和 RLock 的区别以及死锁边界threading.Lock()是普通互斥锁不能重复获取。如果同一个线程已经持有锁又在锁内部再次尝试获取同一把锁程序会阻塞造成死锁。lock threading.Lock() def recursive(n): if n 0: return with lock: recursive(n - 1)这段代码运行到第二层递归时同一个线程再次尝试获取锁但锁没有释放线程会一直等待。如果确实需要在同一线程内重复获取锁可以使用threading.RLock()。RLock 是可重入锁允许同一个线程多次获取。但它也会带来额外复杂度能不用尽量不用。写多线程时还要注意加锁顺序。如果线程 A 先获取锁 1 再获取锁 2线程 B 先获取锁 2 再获取锁 1两个线程就可能互相等待对方释放锁最终死锁。项目里如果有多把锁最好统一加锁顺序或者用更高级的并发模型减少手写多把锁的次数。5. 线程之间传递任务Queue 和线程池更符合工程习惯5.1 为什么不能只靠全局列表存结果新手容易踩的坑是让多个线程写同一个全局列表然后主线程再读这个列表收集结果。results [] def worker(item): results.append(item * 2)如果只是简单append在 CPython 里很多情况碰巧不出错但这不是线程安全的写法。多个线程同时操作一个列表时也可能出现数据覆盖或长度不一致的问题。更稳定的做法是使用queue.Queue。它内部有锁机制专门用于多线程环境下的数据传递。5.2 用 Queue 实现生产者消费者模型生产者负责生成任务消费者负责处理任务。两者通过队列解耦。import queue import threading import time task_queue queue.Queue(maxsize5) stop_event threading.Event() def producer(): for i in range(10): task_queue.put(i) print(f生产者放入任务 {i}) time.sleep(0.05) # 等待所有任务被消费完成 task_queue.join() stop_event.set() def consumer(name): while not stop_event.is_set(): try: item task_queue.get(timeout0.1) except queue.Empty: continue print(f消费者 {name} 处理任务 {item}) # 模拟处理耗时 time.sleep(0.02) task_queue.task_done() producer_thread threading.Thread(targetproducer) consumer_threads [threading.Thread(targetconsumer, args(fC{i},)) for i in range(3)] producer_thread.start() for ct in consumer_threads: ct.start() producer_thread.join() for ct in consumer_threads: ct.join() print(所有任务处理完成)这里的核心点是task_queue.get()会从队列中取出一个任务task_queue.task_done()告诉队列这个任务已经处理完成task_queue.join()会阻塞直到所有放入队列的任务都被调用过task_done()消费者不能直接写成死循环要用stop_event控制退出。如果没有退出条件消费者线程会一直在get()上等待程序结束时可能一直卡住。5.3 用 ThreadPoolExecutor 做更轻量的多线程管理threading.Thread适合理解线程模型但如果任务数量很多比如循环 1000 次创建 1000 个线程手动管理线程列表就不太合适了。频繁创建线程会有系统开销线程太多还可能导致资源占用过高。更推荐的方式是使用concurrent.futures.ThreadPoolExecutor。它底层仍然使用线程但帮你管理线程池、等待任务、获取返回值。from concurrent.futures import ThreadPoolExecutor, as_completed import time import random files [fhttps://example.com/file/{i}.zip for i in range(6)] def download_one(url, index): cost random.uniform(0.2, 0.6) time.sleep(cost) return index, cost with ThreadPoolExecutor(max_workers3) as executor: futures [ executor.submit(download_one, url, index) for index, url in enumerate(files) ] for future in as_completed(futures): index, cost future.result() print(f任务 {index} 完成耗时 {cost:.2f}s)executor.submit()会返回一个 Future 对象代表一个还未执行完毕的任务。as_completed()会按任务完成顺序返回结果而不是任务提交顺序。ThreadPoolExecutor尤其适合批量接口请求和批量文件下载。你不需要手动start()和join()只要设置合理的max_workers然后从 Future 里拿结果就行。这里要提醒新手max_workers不建议一上来就调得很大。实际项目里线程池大小要结合任务类型、对方服务能接受的并发量、本机资源综合判断。盲目设置 100 个线程去请求同一台服务器很可能给对方服务造成压力也可能影响本机稳定性。6. 线程卡住、程序不退出、输出错乱按这个顺序排查6.1 程序卡住或无法退出时先看线程存活状态很多人在本地跑多线程偶尔会遇到程序执行完 main 逻辑后仍不结束。原因往往是还有非守护线程没有退出。非守护线程还在运行主线程代码即使执行到最后一行Python 进程也会等待所有非守护线程结束。如果某个线程里有死循环或者一直阻塞在queue.get()、socket.recv()程序就会看起来像“卡住”。排查的时候可以先打印当前存活的线程import threading print(threading.enumerate())它会输出当前进程里所有线程对象。通过线程名称和状态能大致判断是哪些线程没结束。如果确定某些线程应该一直存在但不应该阻止程序退出可以把它设置成daemonTrue。如果线程是临时任务最好的方式不是设置 daemon而是让线程从任务队列读到一个“退出信号”后主动结束。6.2 结果不正确时不要急着改代码先确认共享资源如果多线程程序最终结果随机变化比如总计数少算、日志缺失、生成文件数量不对首先不要怀疑语法优先检查是不是有共享变量或共享文件。多线程里的“线程安全”是一个很现实的问题。每个线程都在独立执行但共享资源只有一个副本。修改共享资源时必须通过锁、队列或其它同步机制保证数据一致。排查顺序建议是找到共享资源是全局变量、列表、字典还是同一个文件看操作是否非原子、append后可能继续修改、文件写入等看是否所有读写共享资源的线程都加了同一把锁看锁的范围是否覆盖了“读取、修改、写回”整个过程。如果只是读共享数据不加锁通常问题不大。如果涉及到写一定要加锁或者使用线程安全容器。6.3 速度没有提升时要区分任务类型和竞争开销多线程加完之后总耗时没有下降甚至变慢常见原因有这么几个现象常见原因更稳妥的方向耗时没变化任务本身是 CPU 计算被 GIL 限制改用多进程或异步耗时变长加锁范围太大线程大量排队缩小临界区检查锁设计多线程比单线程慢很多创建线程数量过多频繁切换使用线程池限制并发数前几个任务很快后面越来越慢线程在等待共享资源或接口慢慢变慢查看日志耗时分布确认瓶颈运行结果不稳定共享数据没有同步使用 Lock、Queue 或其它并发结构很多问题看起来是“线程没写好”实际上可能是任务类型不合适或者输入数据格式不一致。6.4 异常处理的位置也很重要线程里抛出的异常不会像单线程一样从外层函数暴露出来。如果没有在worker内部捕获异常线程可能静静退出主线程很难知道具体任务发生了什么。所以调试阶段建议尽量在任务函数里加上异常捕获和日志输出def download_one(url, index): try: # 真正处理逻辑 pass except Exception as e: print(f任务 {index} 出错{e})生产环境里更推荐把异常写入日志系统方便后续定位。排查多线程问题时先看日志再改参数不要一上来就调并发数。7. 最后的建议threading 不是唯一解但一定要先把基础模型弄清楚7.1 几种并发方案到底怎么选threading、multiprocessing、asyncio是 Python 里常见的并发方案三者侧重点不同。场景推荐方案原因网络请求、文件下载、接口调用这类 IO 密集型ThreadPoolExecutor或threading代码直观等待时能切换任务纯 CPU 计算、数据分析、算法任务multiprocessing多进程能绕过 GIL利用多核大量高并发 IO比如上万连接asyncio单线程事件循环资源占用更低需要明确资源竞争控制的底层学习threadingLock能深入了解线程同步过程这里不是让你从threading学完后立刻放弃它。只有理解了线程创建、join、daemon、Lock、Queue才能真正明白线程池为什么好用也知道哪些坑需要靠技术方案规避。7.2 我实际写代码时一般怎么控制多线程我会按这个顺序来处理一个多线程任务先写一个单线程版本确保核心逻辑正确统计单次任务耗时判断是 IO 密集还是 CPU 密集如果是 IO 密集先使用小并发数测试比如 3 到 5 个线程检查输出结果和文件产物是否正确确认日志中能看到任务编号、任务开始时间、任务结束时间排到任务稳定后再逐步提高并发数最后考虑失败重试、超时控制、队列上限、输出目录规划。这样操作的好处是一旦出错你能快速判断是核心逻辑问题、线程同步问题还是并发数设置问题。7.3 需要警惕的几个认知多线程不是越多越好。线程数量过多会带来内存占用、上下文切换、锁竞争等问题。多线程不会解决所有“慢”的问题。如果瓶颈在网络带宽、磁盘 IO、第三方服务响应速度就算线程数量加到很大总吞吐也未必明显提升。多线程程序的调试难度比单线程高。你在单线程环境里打印的日志顺序可能是固定的多线程环境下输出顺序会变化。实际项目里要给每个任务一个唯一编号并把关键信息都写进日志不然排错会非常痛苦。也许真正跑完这些例子后你会发现threading的 API 并不难难的是如何判断一个任务到底该不该加线程、加了之后怎么保证数据安全、遇到问题从哪个环节开始查。把这几个问题想清楚多线程基础才算真正掌握。