Python多线程与threading模块实战：从线程创建到GIL锁机制详解

在 Python 进阶路上多线程几乎是绕不开的一道坎。很多人学了函数、类、文件操作之后一旦遇到“下载一批图片”“同时处理多个请求”“爬虫抓取多个页面”这类需求就会发现单线程程序慢得让人着急。第 37 天我们来把 Python 的threading模块彻底讲透从一个线程怎么创建到锁怎么加再到真实场景里到底该不该用多线程全部用代码过一遍。这篇教程会围绕几个核心问题展开Python 里线程和进程到底什么关系、threading模块怎么用、join和daemon是什么、多线程访问共享数据为什么会出问题、Lock该怎么加以及 GIL 对 Python 多线程到底有什么影响。每个知识点都会给出可以直接运行的代码示例建议读者顺手打开自己的 Python 环境跟着敲一遍。文末还会附上多线程面试常见问题和工程实践建议方便对照排查自己的知识盲区。1. 核心知识点速览Python 多线程与 threading 模块主题说明项目定位Python 标准库多线程编程教程核心模块threadingPython 内置无需 pip 安装适用 Python 版本Python 3.x3.6 均可推荐 3.8前置知识Python 基础语法、函数、类、列表字典操作学习目标掌握线程创建、启动、等待、守护线程、锁机制典型应用I/O 密集型任务、网络请求、文件读写、爬虫不适合场景CPU 密集型计算受 GIL 限制建议用 multiprocessing运行平台Windows / Linux / macOS 均可运行这里先给一个结论Python 的threading模块解决的是“并发”问题不是“并行”问题。它适合让程序在等待网络响应、磁盘读写时去执行别的任务从而大幅缩短总耗时。2. 使用场景与学习边界什么时候学多线程什么时候别用多线程不是银弹学之前先搞清楚它能解决什么问题也要知道它不适合解决什么问题。2.1 适合用多线程的场景I/O 密集型任务是 Python 多线程的主场。所谓 I/O 密集型指程序大部分时间都在等待外部设备返回数据CPU 本身反而比较空闲。典型场景包括网络爬虫批量请求网页每个请求要等服务器返回。批量下载图片、文件、视频。从多个 API 接口拉取数据。读写大量小文件磁盘 I/O 频繁但数据量不大。数据库查询量大每次查询都有网络开销。同时运行多个定时任务或监控任务。在这些场景中一个线程发起请求后阻塞等待另一个线程立刻被调度执行程序总耗时可缩短到原来的几分之一甚至几十分之一。2.2 不适合用多线程的场景CPU 密集型任务不适合用 Python 多线程。例如大量数学计算、图像像素循环处理、复杂算法递归、大数据排序等。这类任务需要 CPU 持续高强度工作而 Python 的 GIL 全局解释器锁会让同一时刻只有一个线程在执行 Python 字节码多线程反而可能因为线程切换增加额外开销速度不升反降。CPU 密集场景的正确选择是multiprocessing进程池让多个进程分别占用不同的 CPU 核心。这是第 38 天之后会展开的内容本文先不深入。2.3 学习路径建议建议按以下顺序学完本篇文章弄清进程与线程的区别。学会用threading.Thread创建线程。掌握start()、join()、daemon的使用。理解共享数据的线程安全问题。学会用Lock解决竞态条件。用两个综合例程验证所学。了解 GIL 对多线程的底层影响。最后做一组面试题自测。3. 环境准备Python 多线程编程的前置条件3.1 安装 Pythonthreading是 Python 标准库不需要额外安装。你只需要一个能运行 Python 3 的环境即可。Windows 用户建议直接到 Python 官网下载安装包安装时务必勾选“Add Python to PATH”。如果安装后命令行输入python没有反应说明环境变量没有配置好需要手动把 Python 安装路径加入 PATH。macOS 用户建议使用 Homebrew 安装brew install pythonLinux 用户多数发行版自带 Python 3可直接使用。没有的话执行sudo apt update sudo apt install python3验证环境python --version能输出 Python 3.x 版本号就说明环境没问题。为了代码编辑体验更好建议安装 VS Code 并配置 Python 扩展或者直接使用 PyCharm 社区版。VS Code 里新建.py文件后按CtrlF5即可直接运行。3.2 验证 threading 模块可用在 Python 交互式环境或新建脚本中执行import threading print(threading.__version__) # Python 3.10 不再暴露版本号会报 AttributeError print(threading.current_thread().name)threading模块在 Python 3.x 中已经非常稳定。输出当前线程名称为MainThread即说明模块正常工作。3.3 基础概念准备写多线程代码前先把这几个核心概念搞清楚。进程操作系统分配资源的基本单位。一个程序启动后就是一个进程拥有独立的内存空间。线程进程内部的任务执行单元。一个进程可以包含多个线程线程之间共享进程的内存空间。并发多个任务交替执行看起来像是同时进行实际在单核 CPU 上是通过时间片切换完成的。并行多个任务真的在同一时刻同时执行需要多核 CPU 支持。Python 中的 GIL 决定 CPython 解释器同一时间只允许一个线程执行 Python 字节码。所以 Python 的多线程更像是一种“并发”工具用于在等待 I/O 时切换到其他任务而不是把计算任务分给多个 CPU 核心并行处理。4. threading 模块核心用法与代码实战4.1 最简单的线程创建方式直接实例化 Threadthreading.Thread是threading模块最常用的类。创建线程有两种方式直接传入函数或传入可调用对象。第一种方式把普通函数作为target传入import threading import time def work(): print(f子线程 {threading.current_thread().name} 开始执行) time.sleep(2) print(f子线程 {threading.current_thread().name} 执行完毕) print(f主线程 {threading.current_thread().name} 开始) # 创建线程target 指向要执行的函数 t threading.Thread(targetwork, nameWorkerThread) # 启动线程 t.start() print(主线程继续往下执行)运行这段代码控制台输出顺序可能类似主线程 MainThread 开始 主线程继续往下执行 子线程 WorkerThread 开始执行 等待 2 秒 子线程 WorkerThread 执行完毕注意观察输出顺序调用start()之后主线程并不会阻塞等待子线程结束而是继续往下执行。子线程什么时候真正被调度执行由操作系统的线程调度器决定。你可能会发现WorkerThread 开始执行这一行在主线程继续往下执行之前或之后出现这完全正常。如果你希望主线程等子线程执行完毕后再继续就需要用到join()import threading import time def work(): print(f子线程 {threading.current_thread().name} 开始执行) time.sleep(2) print(f子线程 {threading.current_thread().name} 执行完毕) t threading.Thread(targetwork, nameWorkerThread) t.start() # 等待子线程执行完毕 t.join() print(主线程在子线程结束后继续执行)加上join()后输出顺序变成固定的子线程先完整执行完主线程再打最后一行。4.2 自定义线程类继承 Thread 并重写 run除了直接传入函数还可以通过继承Thread类并重写run()方法创建线程。当线程调用start()时内部会自动调用run()方法。这种方式适合逻辑比较复杂的场景可以把线程相关的属性封装在类中import threading import time class DownloadThread(threading.Thread): def __init__(self, file_name, url): super().__init__() self.file_name file_name self.url url def run(self): print(f[{self.name}] 开始下载 {self.file_name}地址: {self.url}) time.sleep(3) print(f[{self.name}] {self.file_name} 下载完成) if __name__ __main__: tasks [ (图片1.jpg, https://example.com/img1.jpg), (图片2.jpg, https://example.com/img2.jpg), (图片3.jpg, https://example.com/img3.jpg), ] threads [] for name, url in tasks: t DownloadThread(name, url) threads.append(t) t.start() # 等待所有线程执行完毕 for t in threads: t.join() print(所有下载任务已完成)这里self.name是Thread类的内置属性线程未手动命名时系统会自动分配Thread-1、Thread-2这样的名称。4.3 线程生命周期与 daemon 守护线程Python 程序运行逻辑中有个重要规则主线程结束时如果仍有非守护线程存活程序会等待所有非守护线程结束才退出如果只剩下守护线程程序会直接结束。守护线程通过daemonTrue设置。典型应用是后台日志收集、心跳检测、垃圾回收辅助线程等。先看非守护线程的行为import threading import time def work(): print(子线程开始) time.sleep(3) print(子线程结束) t threading.Thread(targetwork) t.start() print(主线程即将结束)输出顺序是子线程开始 主线程即将结束 程序停留约 3 秒 子线程结束因为t不是守护线程主线程执行完后必须等t执行完程序才会结束。再看守护线程import threading import time def work(): print(守护线程开始) time.sleep(3) print(守护线程结束) t threading.Thread(targetwork, daemonTrue) t.start() print(主线程即将结束)输出顺序可能变成守护线程开始 主线程即将结束 程序立即退出守护线程的结束语句很可能根本来不及打印因为主线程结束后程序直接退出不会等守护线程。实际开发中守护线程常用于线程池中的后台清理任务。“当主线程结束守护线程自动销毁”这个特性非常方便但要注意守护线程中的资源清理逻辑不能依赖主线程的返回值。4.4 多线程访问共享数据线程安全问题初现多线程最大的坑不是创建线程而是多个线程同时修改共享数据导致结果错误。来看一个经典例子import threading count 0 def add(): global count for _ in range(1000000): count 1 def sub(): global count for _ in range(1000000): count - 1 t1 threading.Thread(targetadd) t2 threading.Thread(targetsub) t1.start() t2.start() t1.join() t2.join() print(f最终 count {count})如果两个线程严格串行执行count最终应该还是0因为加一百万次、减一百万次相互抵消。但实际多次运行这段代码结果几乎不可能是 0而是一个正数或负数。这就是典型的竞态条件。原因在于count 1这一行代码在 Python 底层不是一步完成的它至少包含三条 CPU 指令读取count当前值。执行加法运算。把新值写回count。线程 A 执行完第 2 步还没写回时线程 B 也可能读取到旧的count值并开始计算。两个线程同时基于同一个旧值做运算最后写回的新值就相互覆盖。很多次的更新操作白白丢失最终结果就偏离了预期。4.5 Lock 锁解决竞态条件threading.Lock提供互斥锁机制保证同一时刻只有一个线程能进入临界区。加锁后的代码段成为“原子操作”其他线程必须等待锁被释放才能进入。改进后的代码import threading count 0 lock threading.Lock() def add(): global count for _ in range(1000000): with lock: count 1 def sub(): global count for _ in range(1000000): with lock: count - 1 t1 threading.Thread(targetadd) t2 threading.Thread(targetsub) t1.start() t2.start() t1.join() t2.join() print(f最终 count {count})输出结果每次都是稳定的0。Lock有两种使用方式方式一手动调用acquire()和release()lock.acquire() try: count 1 finally: lock.release()方式二使用with语句推荐的写法with lock: count 1第二种写法可以保证即使代码抛出异常锁也会被正确释放。日常开发中建议只用with方式。关于锁还有一些细节值得留意。threading.Lock()是可重入的普通锁吗答案是普通Lock不可重入。如果同一个线程在未释放锁的情况下再次调用acquire()程序会死锁。需要可重入锁的话应使用threading.RLock()。另外加锁范围越大并发效率越低。合理做法是只锁定“真正需要保护的那几行代码”不要在整个循环外包一层锁否则两个线程依然会变成串行执行失去多线程的意义。4.6 线程池concurrent.futures 与 ThreadPoolExecutor虽然threading模块是本文核心但必须提前介绍线程池。实际项目里直接手动创建几十个裸线程并不是好方案线程的创建和销毁都有开销大量线程同时运行反而会导致频繁上下文切换CPU 使用率飙升但任务完成速度反而下降。Python 推荐用concurrent.futures.ThreadPoolExecutor管理线程from concurrent.futures import ThreadPoolExecutor import time def fetch_data(url): print(f开始请求: {url}) time.sleep(2) return f{url} 的数据 urls [ https://api.example.com/user/1, https://api.example.com/user/2, https://api.example.com/user/3, https://api.example.com/user/4, https://api.example.com/user/5, ] # 创建最多 3 个线程的线程池 with ThreadPoolExecutor(max_workers3) as executor: futures [executor.submit(fetch_data, url) for url in urls] for future in futures: print(future.result())这段代码中线程池最多同时运行 3 个线程5 个任务按队列依次执行。with块结束时线程池会等待所有任务完成并自动清理资源。比起手动管理Thread对象的start()和join()简洁非常多。4.7 线程间通信 Queue多线程之间交换数据时手工加锁管理共享列表很容易出错。Python 的queue.Queue是线程安全的队列内部自己实现了锁机制专门用于多线程生产者-消费者模型。import threading import queue import time def producer(q): for i in range(5): item f产品-{i} q.put(item) print(f生产了: {item}) time.sleep(1) def consumer(q): while True: item q.get() if item is None: break print(f消费了: {item}) q.task_done() q queue.Queue(maxsize3) threads [ threading.Thread(targetproducer, args(q,)), threading.Thread(targetconsumer, args(q,)), ] for t in threads: t.start() q.join() print(所有任务完成)生产者线程向队列添加数据消费者线程从队列取出数据。Queue内部已经加锁多线程同时put或get不会出现数据错乱比手动维护list加Lock可靠得多。5. 综合代码实战批量下载任务中的多线程应用为了把前面所有知识点串起来这里设计一个贴近真实业务的下载任务案例。假设需要在 10 秒内从模拟服务器下载 6 个文件单线程需要约 18 秒用多线程可以把耗时压缩到约 6 秒。import threading import time import random class MockDownloader: 模拟从网络下载文件的类 def __init__(self): self.lock threading.Lock() self.completed 0 def download(self, file_id): # 模拟不同文件大小导致的不同下载时长 cost random.uniform(2, 4) time.sleep(cost) with self.lock: self.completed 1 print(f文件 {file_id} 下载完成耗时 {cost:.2f}s总进度 {self.completed}/6) def run(self): threads [] for i in range(1, 7): t threading.Thread(targetself.download, args(i,), namefDownload-{i}) threads.append(t) t.start() for t in threads: t.join() print(f全部下载完成共 {self.completed} 个文件) if __name__ __main__: start time.time() MockDownloader().run() print(f总耗时: {time.time() - start:.2f} 秒)这个案例体现了三个重点多个线程同时发起“下载”、用threading.current_thread().name或自定义线程名区分日志、用Lock保护共享进度值避免打印统计错乱。如果这里不用锁self.completed 1在极端情况下也会出现计数丢失的问题。虽然 CPython 对整数自增有一定优化但在高并发场景下依然不可靠工程上默认加锁是最稳妥的。6. 多线程性能观察与 GIL 影响分析6.1 用实验证明 I/O 密集型任务的多线程加速写一个模拟 I/O 等待的程序对比单线程和多线程耗时import threading import time def io_task(): time.sleep(2) # 单线程执行 5 次 start time.time() for _ in range(5): io_task() print(f单线程耗时: {time.time() - start:.2f}s) # 多线程执行 5 次 start time.time() threads [] for _ in range(5): t threading.Thread(targetio_task) threads.append(t) t.start() for t in threads: t.join() print(f多线程耗时: {time.time() - start:.2f}s)运行结果通常非常接近单线程耗时: 10.01s 多线程耗时: 2.01s5 个线程同时睡眠等待总耗时不叠加说明多线程成功利用了 I/O 等待时间。6.2 CPU 密集型任务不要用多线程换一个 CPU 密集型实验import threading import time def cpu_task(): total 0 for i in range(50000000): total i return total start time.time() for _ in range(4): cpu_task() print(f单线程耗时: {time.time() - start:.2f}s) start time.time() threads [] for _ in range(4): t threading.Thread(targetcpu_task) threads.append(t) t.start() for t in threads: t.join() print(f多线程耗时: {time.time() - start:.2f}s)运行后会发现多线程版本很可能比单线程还慢或者只是持平。原因就是 GIL。CPython 解释器中任何时刻只允许一个线程执行 Python 字节码CPU 计算任务无法利用多核优势。线程越多切换开销越大性能反而下降。CPU 密集型任务应改用多进程。这是官方推荐、社区公认的做法。具体代码会在后续文章介绍multiprocessing时详细展开本文先记住结论I/O 密集型用threadingCPU 密集型用multiprocessing。6.3 如何观察线程运行状态线程不是越多越好。观察线程数量最简单的方式import threading import time def work(): time.sleep(3) threads [] for i in range(10): t threading.Thread(targetwork) threads.append(t) t.start() print(f当前活动线程数: {threading.active_count()}) print(f当前线程列表: {threading.enumerate()})threading.active_count()返回当前存活线程数量threading.enumerate()列出所有线程对象。调试时可以通过打印线程名区分不同线程的输出import threading def func(): print(f{threading.current_thread().name} 正在执行) t threading.Thread(targetfunc, nameMyWorker) t.start()线程名最好在创建时就指定这样日志可读性会好很多。7. 线程安全与常见陷阱排查7.1 直接修改共享 list 或 dict多个线程同时append到同一个list、或同时update同一个dict看起来安全但较大数据量下仍可能出现状态不一致。不要赌 CPython 的原子性使用锁保护共享容器是更严谨的做法。7.2 忘记 join 导致数据还没处理完程序就退出核心工作线程是普通线程主线程忘记join()时主线程虽然不会立即退出但某些逻辑如后续统计汇总会提前执行。建议每个start()都有对应的join()不要省略。7.3 死锁锁顺序不一致多个锁存在时不同线程按不同顺序获取锁可能导致互相等待。例如线程 A 持有锁 1 等待锁 2线程 B 持有锁 2 等待锁 1程序永久卡住。避免死锁的方法是让所有线程按相同顺序获取锁或者尽量用单把锁控制临界区。实际业务中要避免锁嵌套锁的范围越小越好。7.4 无限制创建线程导致资源耗尽创建大量线程而不复用每个线程都占用内存和系统资源最终导致 OOM 或线程创建失败。生产环境统一用ThreadPoolExecutor(max_workersN)限流不建议直接循环Thread。7.5daemon线程中的资源清理如果守护线程持有文件句柄或数据库连接主线程退出时资源可能无法正常释放。因此守护线程不要承担需要完整清理的任务宁愿用join()等待或独立的进程服务。8. 实战ThreadPoolExecutor 批量 API 数据拉取手动管理十几个Thread已经让人头疼真实项目中更推荐直接用线程池。这里给出适合做数据采集任务的通用模板from concurrent.futures import ThreadPoolExecutor, as_completed import time import random def fetch_one(user_id): # 模拟调用外部 API cost random.uniform(0.5, 2) time.sleep(cost) return {user_id: user_id, status: ok, delay: round(cost, 2)} user_ids list(range(1, 11)) # max_workers 一般设置为 I/O 型的 5~20 之间需根据目标和API速率调整 with ThreadPoolExecutor(max_workers5) as executor: future_map {executor.submit(fetch_one, uid): uid for uid in user_ids} for future in as_completed(future_map): uid future_map[future] try: result future.result() print(f用户 {uid}: {result}) except Exception as exc: print(f用户 {uid} 请求失败: {exc})重点看两点as_completed按任务完成顺序返回结果不必等待前面的慢任务可以直接把完成的数据写入文件或数据库。future.result()会抛出任务内部的异常调用方必须捕获否则一个任务出错可能导致整个主线程中断。真实采集数据时建议配合重试机制和日志输出。比如单个请求失败时先记录下来全部请求结束后统一补拉。还可以在函数内部加入简单的指数退避重试import time import random def fetch_with_retry(user_id, retries3): for attempt in range(retries): try: # 假设这里是真实的 HTTP 请求 return {user_id: user_id, status: ok} except Exception as exc: wait_time 2 ** attempt random.uniform(0, 1) time.sleep(wait_time) return {user_id: user_id, status: failed}9. Python 多线程面试高频问题与解答思路结合前面代码把面试里常问的几个问题一并整理。这些问题也适合用来检验自己是否真正理解多线程。9.1 Python 的 GIL 是什么GIL 是 CPython 解释器中的全局解释器锁它保证同一时刻只有一个线程在执行 Python 字节码。多线程无法利用多核 CPU 并行执行 CPU 密集型任务。I/O 密集型任务不受影响因为等待 I/O 时会释放 GIL。9.2 GIL 可以被拆除吗Python 官方多次尝试移除 GIL但由于 C 扩展模块大量依赖 GIL 做线程安全保护移除后会影响兼容性和单线程性能。Python 3.13 中提出了自由线程模式但通用生态尚未完全迁移。实际使用中CPU 密集场景仍然推荐多进程。9.3 threading 和 multiprocessing 怎么选线程共享内存创建开销小适合 I/O 密集型进程内存隔离创建开销大但能利用多核适合 CPU 密集型。守护线程和队列在线程间通信更方便进程通信则需要更显式的机制。9.4 什么是竞态条件多个线程同时读写同一个数据资源最终结果取决于线程调度顺序称为竞态条件。9.5 Lock 和 RLock 区别普通Lock只能被 acquire 一次不能重入同一个线程再次 acquire 会死锁。RLock允许同一个线程多次获取锁内部有计数器释放时也需要对应次数适合递归调用场景。9.6 如何安全地在多线程中共享数据优先使用queue.Queue这种线程安全容器访问共享数据时必须加锁。不要依赖“运气”即使某段代码在小数据量测试中没问题大流量下也可能暴露并发 bug。10. Python 多线程学习建议与下一步方向多线程代码能不能写对很大程度取决于对共享数据和锁模型的理解。前几次运行出现奇怪结果非常正常建议刻意做一个实验把count 1的循环次数分别设为 100 次、1 万次、100 万次观察错误概率变化。这个实验能直观感觉到数据量增大后竞态问题更容易出现。日常开发中能不用裸线程就不要用裸线程。ThreadPoolExecutor和线程安全队列几乎能满足绝大多数需求。线程方案如果仍然搞不定高并发 I/O可以再考虑协程也就是asyncio方案。协程在单线程里通过事件循环实现高并发上下文切换开销比线程还低适合大量网络请求但编程思维和线程明显不同可以放在后续学习计划里。100 天精通 Python 系列进行到多线程意味着你已经具备函数、模块、文件类和基础脚本能力开始进入真正的工程并发领域。今天把threading模块中不能出 bug 的部分全部掌握代码全部跑通接下来不管是学网络编程还是爬虫底气都会完全不一样。建议把本文的代码段单独建一个threading_demo目录保存后续复习时直接运行对比实验结果。