很多 Python 初学者学到多线程时，都会抱着这样一个期待：开了多个线程，程序执行速度应该能翻倍吧？然后写一段死循环测试，却发现结果出乎意料——不仅没变快，有时候反而更慢了。于是网上开始流传另一种声音：Python 多线程就是假的，完全没有用，别学了。
这两种说法其实都只对了一半。Python 的threading模块不是万能的加速器，但也不是一个该被抛弃的鸡肋。它真正擅长解决的是 I/O 密集型任务，比如网络请求、文件读写、数据库查询；而在 CPU 密集型计算场景下，它会受到 GIL（全局解释器锁）的限制。如果你正在写爬虫、做批量接口调用、处理大量文件，或者想在 GUI 程序里避免界面卡死，threading都是非常实用的工具。
这篇文章不从概念堆砌开始，而是直接告诉你：多线程适合解决什么问题、不适合解决什么问题，然后从threading模块的核心 API 出发，用代码演示三种创建线程的方式、线程锁、事件、队列、线程池，以及实际工程中常见的坑和排查思路。读完你可以照着代码跑一遍，并且能判断你自己的项目到底该不该用多线程。
1. 为什么你写的 Python 多线程没有变快
先回到开头的困惑：为什么多线程没有让代码更快？
这里必须先解释一个 Python 特有的机制——GIL，全称 Global Interpreter Lock，也就是全局解释器锁。CPython 解释器在同一个进程内，同一时刻只允许一个线程执行 Python 字节码。这意味着，如果你有 8 个线程在跑 CPU 密集型计算，它们并不会真正地同时运行在 8 个 CPU 核心上，而是轮流抢同一把锁。
# 一个直观的类比 单线程进程：食堂只有一个打饭窗口 多线程进程：食堂还是只有一个打饭窗口，只是排队的人分成 8 列 多进程方案：开了 8 个窗口，每列人同时打饭所以，如果是纯计算任务，比如循环求和、图片像素处理、大数据量的数学运算，多线程反而会因为线程创建、上下文切换、锁竞争带来额外开销，表现不如单线程。这种场景下，真正合适的是multiprocessing多进程，或者直接用 NumPy 这类能把计算下沉到 C 层的库。
但 I/O 密集型任务就完全不同。当线程执行requests.get()、file.read()、time.sleep()这类操作时，线程会进入等待状态，此时 GIL 会被释放，其他线程就能继续执行。也就是说，多线程在等待网络响应或磁盘读写的时间里，可以让其他线程去干活，从而实现“并发等待”，整体耗时会大幅缩短。
概括成一张判断表：
| 任务类型 | 特点 | 推荐方案 | 原因 |
|---|---|---|---|
| I/O 密集型 | 网络请求、文件读写、数据库查询 | threading 或 asyncio | 等待期间自动让出 GIL，并发效果明显 |
| CPU 密集型 | 数值计算、循环处理、压缩解压 | multiprocessing | 绕开 GIL，真正利用多核 |
| 混合型 | 既有计算又有等待 | threading + 计算下沉到 C 库 | 提高吞吐，计算部分交给更合适的工具 |
这篇文章后面说的所有示例，默认都是面向 I/O 密集型场景。
2. threading 模块的核心概念与适用场景
2.1 线程与主线程
一个 Python 程序启动后，默认会有一个执行流，这个执行流叫主线程。使用threading模块创建的额外执行流，称为子线程。
每个线程都有自己独立的方法调用栈，但共享同一个进程内的全局变量和内存空间。这是多线程和多进程最大的差异：多线程之间通信成本低，但需要处理共享数据的竞争问题。
看一个最简单的最小示例：
# 文件路径：demo_basic_thread.py import threading import time def worker(): print(f"子线程开始: {threading.current_thread().name}") time.sleep(1) print(f"子线程结束: {threading.current_thread().name}") if __name__ == "__main__": print(f"主线程开始: {threading.current_thread().name}") t = threading.Thread(target=worker, name="worker-thread") t.start() t.join() print(f"主线程结束: {threading.current_thread().name}")运行结果：
主线程开始: MainThread 子线程开始: worker-thread 子线程结束: worker-thread 主线程结束: MainThread这里有两个容易被忽略的细节：
第一，target指定线程要执行的函数，name给线程命名，方便日志排查。
第二，t.join()会让主线程等待子线程执行完毕再继续。如果去掉join()，主线程不会等子线程，程序可能直接执行完退出，子线程还没来得及完整输出。
初学者可以先把线程理解成“程序里另开的几条小流水线”。主线程负责统筹，子线程负责干活，join()就是“等这条流水线收工”。
2.2 守护线程
线程有一个布尔属性daemon，中文叫守护线程。daemon=True的线程不会阻止主线程退出。主线程结束时，守护线程会被强制终止。
默认情况下，threading.Thread创建的非守护线程，主线程会等待它执行完才退出。
# 文件路径：demo_daemon.py import threading import time def background_task(): while True: print("后台任务运行中...") time.sleep(0.5) if __name__ == "__main__": t = threading.Thread(target=background_task, daemon=True) t.start() time.sleep(2) print("主线程即将退出，守护线程会被强制结束")注意：守护线程通常用于日志轮转、心跳检测、自动保存这类“后台服务型”任务。业务中重要的任务不要设置成守护线程，否则主线程退出后任务可能执行到一半被强杀。
2.3 线程生命周期
一个线程从创建到结束，大致经历以下状态：
| 状态 | 说明 | 触发方式 |
|---|---|---|
| 创建 | Thread 对象被实例化 | threading.Thread(target=fn) |
| 就绪/运行 | 线程开始执行 | t.start() |
| 阻塞/等待 | 线程等待 I/O 或锁 | 遇到time.sleep()、queue.get() |
| 死亡 | 函数执行完毕或发生未捕获异常 | 自动结束或异常退出 |
threading模块没有直接提供获取线程运行状态的方法，但可以通过t.is_alive()判断线程是否还在运行。
3. 环境准备与实验说明
本文的所有代码基于 Python 3.10 版本编写，但涉及的 API 在 Python 3.6 之后的版本都能正常使用。
如果你的电脑还没装 Python，或者不确定当前版本，可以先执行下面的命令检查：
python --version建议在项目目录下创建一个独立的虚拟环境，避免依赖包污染系统环境：
# Windows python -m venv venv venv\Scripts\activate # Linux / macOS python3 -m venv venv source venv/bin/activate为了让后续的示例效果更明显，建议安装requests库，用于网络请求场景的演示：
pip install requests如果你使用的是 Anaconda，它自带requests，直接跳过这一步即可。
本文的核心代码不依赖第三方库的只有前三节；从第 5 节网络请求示例开始才需要requests。
4. 三种创建线程的常见方式
很多教程一上来就讲多线程的完整架构，其实没必要。你先掌握三种创建线程的方式，后续再谈复杂的线程协作。
4.1 方式一：直接通过 Thread 类传入函数
这是最常用、也最推荐新手使用的方式。
# 文件路径：demo_create_thread.py import threading import time def fetch_url(url): print(f"正在请求: {url}") time.sleep(1) print(f"请求完成: {url}") if __name__ == "__main__": urls = [ "https://example.com/api/1", "https://example.com/api/2", "https://example.com/api/3", ] threads = [] for url in urls: t = threading.Thread(target=fetch_url, args=(url,)) threads.append(t) t.start() for t in threads: t.join() print("所有请求已完成")关键点在于args=(url,)。注意这里必须是一个元组，如果写成args=(url)，就等于传入一个字符串，Python 会报TypeError。当只有一个参数时，一定要记得加结尾逗号。
4.2 方式二：继承 threading.Thread 重写 run 方法
这种方式适合业务逻辑比较复杂、需要在线程中保存状态的场景。
# 文件路径：demo_subclass_thread.py import threading import time class DownloadThread(threading.Thread): def __init__(self, task_id): super().__init__() self.task_id = task_id self.result = None def run(self): print(f"任务 {self.task_id} 开始") time.sleep(1) self.result = f"任务 {self.task_id} 的返回值" print(f"任务 {self.task_id} 结束") if __name__ == "__main__": threads = [] for i in range(1, 4): t = DownloadThread(i) threads.append(t) t.start() for t in threads: t.join() for t in threads: print(t.result)这段代码里有一个非常实用的设计：把执行结果保存在self.result上。join()之后，主线程可以从线程对象中取出子线程的运行结果。这比使用全局变量来收集结果要清晰得多。
继承方式的好处是代码组织性好，每个线程相当于一个具有内部状态的任务对象；缺点是写法略冗余。简单任务用方式一就够了。
4.3 方式三：ThreadPoolExecutor 线程池
每次任务都手动创建线程，在线程数量大时效率不高。更工程化的做法是使用线程池：预先创建一批线程，任务来了就分配一个空闲线程执行。
concurrent.futures.ThreadPoolExecutor是 Python 内置的线程池实现：
# 文件路径：demo_thread_pool.py from concurrent.futures import ThreadPoolExecutor, as_completed import time def handle_task(task_id): time.sleep(1) return f"任务 {task_id} 处理完成" if __name__ == "__main__": task_ids = list(range(1, 11)) with ThreadPoolExecutor(max_workers=3) as executor: future_map = {executor.submit(handle_task, task_id): task_id for task_id in task_ids} for future in as_completed(future_map): task_id = future_map[future] try: print(future.result()) except Exception as e: print(f"任务 {task_id} 执行失败: {e}")运行结果（顺序可能不同）：
任务 1 处理完成 任务 3 处理完成 任务 2 处理完成 任务 5 处理完成 ... 任务 10 处理完成这里解释几个常用方法：
executor.submit(fn, *args)提交一个任务到线程池，返回一个Future对象。future.result()获取任务返回值，如果任务抛异常，这里会重新抛出异常。as_completed(futures)是一个迭代器，哪个任务先完成就先返回哪个 future，适合处理执行时间不定的任务。with语句退出时会调用executor.shutdown()，等待所有已提交任务执行完毕。
线程池真正解决了“线程数量管理”的问题。如果不限制线程数，你就创建 1 万个线程去请求接口，操作系统资源会被快速耗尽。合理设置max_workers可以保护系统资源。
5. 线程同步：多线程不是各跑各的
线程之间共享全局变量，这会带来一个经典问题：多个线程同时修改一个数据，最终结果不符合预期。
5.1 不加锁的竞态条件
先看一个经典的计数器示例：
# 文件路径：demo_race_condition.py import threading counter = 0 def increment(): global counter for _ in range(1000000): counter += 1 if __name__ == "__main__": threads = [] for _ in range(10): t = threading.Thread(target=increment) threads.append(t) t.start() for t in threads: t.join() print(f"counter 的最终值: {counter}")如果程序按顺序正确执行，10 个线程各自加 100 万次，结果应该是 10000000。但实际运行多次，结果往往是几百万、几十万，每次都不一样。
原因在于counter += 1并不是一个原子操作。它内部可以拆成三步：
- 读取 counter 当前值。
- 计算 counter + 1。
- 把新值写回 counter。
假设两个线程同时读取到 counter = 100，线程 A 算出了 101，线程 B 也算出了 101，然后各自写回，counter 最终只变成 101，而不是 102。这就丢失了一次更新。
5.2 加锁解决竞态条件
threading.Lock可以保证同一时刻只有一个线程能进入临界区：
# 文件路径：demo_lock.py import threading counter = 0 lock = threading.Lock() def increment(): global counter for _ in range(1000000): with lock: counter += 1 if __name__ == "__main__": threads = [] for _ in range(10): t = threading.Thread(target=increment) threads.append(t) t.start() for t in threads: t.join() print(f"counter 的最终值: {counter}")这次每次运行结果都应该是 10000000。
with lock:是 Python 上下文管理器语法，等价于：
lock.acquire() try: counter += 1 finally: lock.release()第二种写法也不能算错，但with语法更安全。如果临界区中间的代码抛出异常，finally还能保证锁被释放；直接调用acquire()后忘记release()，就会造成死锁。
5.3 可重入锁 RLock
Lock 还有一个明显的限制：同一个线程不能连续多次acquire()。如果代码里存在嵌套加锁，或者一个方法内部调用另一个也加锁的方法，就可能死锁。
这时需要使用threading.RLock（可重入锁）。它允许同一个线程多次获取锁，内部用一个计数器记录获取次数，只在全部release()后才真正释放锁。
# 文件路径：demo_rlock.py import threading lock = threading.RLock() counter = 0 def inner(): with lock: global counter counter += 1 def outer(): with lock: print("外层加锁") inner() print("内层调完，counter =", counter) if __name__ == "__main__": outer()如果这里用的是普通Lock，在outer()中获取锁后，再调用inner()尝试获取同一把锁，程序会直接卡死，因为同一线程无法获取自己已经持有的 Lock。RLock就能避免这个问题。
5.4 实际场景：多线程写入同一个文件
写日志是多线程中很常见的场景。如果多个线程同时往一个文件里写，会出现日志内容交错、行被拆断的问题。给写入操作加一把写锁是最直接的解决方案。
# 文件路径：demo_file_write.py import threading import time write_lock = threading.Lock() def write_log(filename, thread_name, content): time.sleep(0.01) with write_lock: with open(filename, "a", encoding="utf-8") as f: f.write(f"[{thread_name}] {content}\n") if __name__ == "__main__": threads = [] for i in range(10): t = threading.Thread(target=write_log, args=("app.log", f"T{i}", f"line {i}")) threads.append(t) t.start() for t in threads: t.join() print("日志写入完成")这里的time.sleep(0.01)故意模拟日志内容的生成耗时。加上写锁后，同一时刻只有一个线程在操作文件，写入内容就是完整的行。
6. 线程之间的通信与协作
锁解决了“数据竞争”，但很多场景下，线程之间还需要互相通知和传递数据。
6.1 使用 Queue 安全传递任务
最推荐的线程间通信方式，是使用queue.Queue。它内部已经实现了线程安全，put()和get()方法都自带了锁机制，不需要你再去加一把锁。
# 文件路径：demo_queue.py import queue import threading import time import random task_queue = queue.Queue() result_queue = queue.Queue() def producer(): for i in range(10): task = f"任务-{i}" task_queue.put(task) print(f"生产: {task}") time.sleep(random.uniform(0.1, 0.3)) def worker(worker_id): while True: try: task = task_queue.get(timeout=3) except queue.Empty: break print(f"线程 {worker_id} 处理: {task}") time.sleep(random.uniform(0.2, 0.5)) result_queue.put(f"{task} 已完成") task_queue.task_done() if __name__ == "__main__": producer_thread = threading.Thread(target=producer) producer_thread.start() workers = [] for i in range(3): t = threading.Thread(target=worker, args=(i,)) workers.append(t) t.start() producer_thread.join() for t in workers: t.join() task_queue.join() print("所有任务处理完毕")这段代码模拟了一个典型的生产者消费者模型：
- 生产者线程往
task_queue里放任务。 - 多个工作线程从队列里取任务并处理。
- 处理结果放入
result_queue，之后你可以用另一个消费者线程来收集结果。
关键方法说明：
| 方法 | 作用 |
|---|---|
put(item) | 向队列添加元素，如果队列满则阻塞 |
get() | 从队列取出元素，如果队列空则阻塞 |
get_nowait() | 非阻塞获取元素，空队列时抛出queue.Empty |
task_done() | 告诉队列当前任务已处理完 |
join() | 阻塞直到队列里所有任务都被标记为task_done |
get(timeout=3)是为了避免工作线程在队列清空后一直阻塞。当队列空了，且 3 秒内没有新任务，线程会捕获queue.Empty后退出。如果不设超时，get()会永远阻塞下去，程序无法正常结束。
6.2 使用 Event 做线程间通知
threading.Event适合从一个线程向另一个线程发送“信号”。它内部维护一个布尔标志，set()会把它变为 True，wait()会阻塞直到标志变为 True。
# 文件路径：demo_event.py import threading import time start_event = threading.Event() def worker(worker_id): print(f"线程 {worker_id} 已就绪，等待启动信号...") start_event.wait() print(f"线程 {worker_id} 收到信号，开始工作") if __name__ == "__main__": threads = [] for i in range(3): t = threading.Thread(target=worker, args=(i,)) threads.append(t) t.start() time.sleep(1) print("主线程发送启动信号") start_event.set() for t in threads: t.join()运行结果：
线程 0 已就绪，等待启动信号... 线程 1 已就绪，等待启动信号... 线程 2 已就绪，等待启动信号... 主线程发送启动信号 线程 0 收到信号，开始工作 线程 2 收到信号，开始工作 线程 1 收到信号，开始工作Event 常用于以下场景：
- 多个工作线程初始化后，先等待主线程统一发出“开始”信号。
- 工作线程需要周期性等待某个外部配置更新的信号。
- 需要优雅关闭线程时，通过 Event 通知循环线程退出。
7. 完整案例：用多线程并发请求接口
把前面的知识点综合起来，做一个贴近真实工作的案例：批量请求一组 URL，并使用线程池管理并发度，最后返回每个请求的状态和耗时。
# 文件路径：demo_concurrent_requests.py import threading import time import requests from concurrent.futures import ThreadPoolExecutor, as_completed # 仅用于演示，实际请求需要保证目标域名允许访问 URLS = [ "https://www.example.com", "https://www.python.org", "https://docs.python.org", ] lock = threading.Lock() results = [] def fetch_one(url): start = time.time() try: resp = requests.get(url, timeout=5) status = resp.status_code length = len(resp.text) except Exception as exc: status = str(exc) length = 0 cost = round(time.time() - start, 2) # 多线程同时向列表追加数据可能造成竞争，这里加一个锁保护 with lock: results.append((url, status, length, cost)) return url, status, cost if __name__ == "__main__": start_all = time.time() with ThreadPoolExecutor(max_workers=3) as executor: futures = [executor.submit(fetch_one, url) for url in URLS] for future in as_completed(futures): url, status, cost = future.result() print(f"完成: {url}, 状态: {status}, 耗时: {cost}s") total_cost = round(time.time() - start_all, 2) print(f"总耗时: {total_cost}s") print("\n结果汇总:") for url, status, length, cost in results: print(f"{url} -> 状态码: {status}, 内容长度: {length}, 耗时: {cost}s")注意，requests.get()是阻塞式网络请求。当线程发出网络请求后，它会在等待响应时释放 GIL，所以多线程能够明显缩短整体请求时间。
如果所有请求是串行的，N 个请求的总耗时约为 N 个请求耗时之和；使用线程池后，总耗时约等于最慢的那个请求耗时，前提是线程池大小足够。
实际工程中，线程池的max_workers建议结合目标服务器的承受能力来设置，并不是越大越好。大量并发请求可能触发对方限流或造成 IP 被临时封禁。合规爬虫应当控制请求频率，并遵守目标网站的robots.txt和访问协议。
8. 常见问题与排查思路
多线程程序的调试难度比单线程高很多。遇到问题不要盲改代码，先按下面的表格逐项排查。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 程序运行结束后线程还没执行完 | 主线程未调用join() | 检查代码中是否对每一个Thread调用了join() | 在main最后统一join()所有线程 |
| 多个线程同时修改同一个列表/字典，数据丢失 | 共享数据写入存在竞态条件 | 在写入位置打印线程名和当前数据长度 | 使用threading.Lock保护写入，或改用queue.Queue |
| 程序卡死，无法退出 | 死锁或某个线程阻塞在get()上 | 使用py-spy dump查看线程栈，或打印每个线程的启动状态 | 避免嵌套锁，使用RLock，给queue.get()设置超时 |
| CPU 使用率始终只有 100% 左右 | 纯 CPU 计算任务受 GIL 限制 | 观察任务类型是计算密集还是 I/O 密集 | 改用multiprocessing或把计算逻辑改用 NumPy |
| 线程函数抛异常，但主线程看不到 | 子线程异常默认不会传播到主线程 | 在run()方法外层捕获异常并打印 traceback | 自定义线程基类，统一捕获并记录异常 |
| 线程池任务执行顺序和提交顺序不一致 | 并发任务本来就是无序完成 | 打印各任务完成时间 | 需要严格有序时改用单线程或按索引收集结果 |
daemon=True的线程任务没执行完程序就退出 | 守护线程不阻止主线程退出 | 确认线程任务是否具备事务性 | 非守护任务不要设置daemon=True |
max_workers设置很大，但速度没有提升 | I/O 服务和本地资源存在瓶颈 | 检查目标服务响应时间、本机文件句柄数 | 适当降低线程数，或调整系统文件句柄限制 |
再补充两个排查工具和技巧：
第一，使用threading.enumerate()打印当前存活线程列表：
import threading for thread in threading.enumerate(): print(f"线程名: {thread.name}, 是否存活: {thread.is_alive()}")第二，使用 Python 自带的日志模块记录线程名。logging默认支持%(threadName)s格式，能在日志中直接看到是哪条线程打印的信息：
import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(threadName)s] %(message)s' ) logging.info("这条日志会显示线程名")排查死锁时，这两个信息通常能快速定位问题。
9. 最佳实践与工程建议
多线程代码写起来容易，写对很难。这里总结几条我在实践中认为最重要的工程建议。
优先使用线程池，而不是手动创建大量线程。
ThreadPoolExecutor帮你管理线程生命周期，减少线程创建和销毁的开销，同时能限制最大并发数量。手动创建线程适合线程数量很少、场景简单的程序。共享数据结构统一通过加锁或 Queue 访问。 不要把“往字典里加一条数据”当成一个绝对安全的小操作。Python 对于单个赋值操作在 CPython 下往往带有 GIL 保护，但复合操作（如
list.append前先判断长度）仍然可能出错。最稳妥的方式是：所有跨线程的共享状态变更都加锁，或者使用线程安全的queue.Queue。锁的粒度要尽可能小。 锁住的代码范围越小，线程等待时间越短，并发效率越高。比如有 1000 万次循环，不要在循环外面一次性加锁然后循环 1000 万次，那样等于把多线程又变成串行了。正确做法是每次操作前加锁，操作完立刻释放。不过频繁加锁也会带来性能消耗，需要在实践中找到平衡。
使用超时和取消机制。 队列取任务的
get(timeout=...)、等待锁的acquire(timeout=...)、等待线程结束的join(timeout=...)，这些超时参数在异常场景下非常重要。生产环境中，一个线程可能因为外部服务无响应而卡死，超时机制能保证程序不会永久阻塞。异常必须在线程内部捕获。 线程函数里抛出的异常，不会自动传递给主线程。如果你想知道哪个线程发生了什么错误，必须在子线程内部捕获并记录。更好的做法是在线程函数的入口写统一的
try/except并调用日志模块。不要为了用多线程而用多线程。 任务总数很少，比如只有两个请求，单线程串行可能只需要 0.5 秒，用线程池反而要花时间创建线程。使用多线程前，先量化任务的等待时间和数量。如果任务本身很快，并发带来的收益可以忽略不计。
注意线程本身的启动开销。 线程创建不是零成本的。如果任务是百万级别的短任务，每条线程又很轻量，更适合使用线程池复用线程。线程数量超过几千以后，线程调度本身会成为瓶颈。
10. 总结与后续学习方向
回到最开始的问题。Python 的threading模块不是让你把 CPU 计算加速到多核并行的银弹，它是 Python 在 I/O 密集型场景下提升吞吐量的实用工具。理解 GIL 的边界，是理解 Python 多线程的前提。在 I/O 等待场景下，threading可以让多个请求并发等待，大幅缩短总耗时；在 CPU 密集计算场景下，应该选择multiprocessing或把核心计算交给 C 扩展库。
本文通过完整代码演示了三种创建线程的方式、Lock 和 RLock 的用法、Queue 如何做线程间通信、Event 如何传递信号，以及线程池在生产场景中的标准写法。这些知识点放在一个真实的批量请求任务中串起来之后，你已经具备在 Python 爬虫、后端服务、自动化脚本里接入多线程的基本能力。
下一步你可以从两个方向继续深入：
- 一个是向异步编程方向探索，Python 的
asyncio在单线程内使用事件循环处理大量连接，在高并发 I/O 场景下是另一种解决方案。多线程和多进程适合并发执行多个独立任务，而异步适合大量任务之间快速切换。 - 另一个是向多进程方向探索，学习
multiprocessing模块、进程池、进程间通信，理解 Python 中进程与线程的选择边界。
建议你不只是在本地运行示例代码，而是找一个真实的小项目做实验。找一个需要并发请求的接口列表，用单线程、多线程、线程池分别测一遍总耗时和资源占用，你就对 Python 并发模型有了更直观的体感。欢迎收藏这篇文章，后续遇到线程相关问题，可以回来对照排查表快速定位。
