这次我们来看 Python 多线程里的threading模块。很多人在写 Python 脚本时习惯用for循环一个任务一个任务排队执行比如批量下载文件、批量请求接口、批量处理图片只要任务一多程序的耗时基本就是“任务数量 x 单个任务时间”。threading模块解决的就是这一类问题它提供了一套标准 API让你可以在一个进程里同时启动多个线程把相互独立的任务并行推进。先明确两个前提第一今天所有代码都基于 Python 标准库只要解释器版本在 3.8 以上不需要装额外依赖就能直接跑第二Python 多线程受 GIL全局解释器锁影响不是所有场景都能加速。所以这篇文章不只是讲“怎么创建线程”更会讲清楚什么场景适合用、什么场景不适合用以及共享变量竞争、线程锁、队列通信、线程池批量任务这些高频需求怎么落地。如果你当前正处于“会用requests和for循环但觉得批处理太慢”的阶段今天的内容会直接解决你的问题。文章会按“环境准备 - 线程创建 - 生命周期控制 - 锁 - 队列 - 线程池 - 性能观察 - 排错 - 最佳实践”的顺序把threading模块真正讲透。1. threading 模块核心能力速览在动手写代码前先用一张表快速建立对threading模块的整体认知能力项说明模块类型Python 标准库无需额外安装核心类Thread、Lock、RLock、Event、Semaphore配合标准库queue.Queue主要功能创建线程、启动线程、等待线程结束、守护线程、线程锁、线程间通信、生产者消费者模型适用任务网络请求、文件读写、数据库查询等 IO 密集型任务不适合任务纯 CPU 计算密集型任务例如大量浮点运算、复杂循环计算限制原因CPython 解释器的 GIL 使得同一时刻只有一个线程执行 Python 字节码替代方案CPU 密集用multiprocessing/ProcessPoolExecutor高并发 IO 可用asyncio工程化推荐管理大量任务时优先使用concurrent.futures.ThreadPoolExecutor上手难度低理解线程生命周期和共享变量问题即可这里的核心结论是threading很轻、很直接但它不是“无脑提速工具”。你用它处理网络请求和文件任务效果立竿见影如果拿它去加速一个纯计算的循环很可能不仅不加速反而因为上下文切换变慢。2. Python 多线程的适用边界与真实场景很多初学者看到“多线程”三个字会以为它等于“多个任务同时执行”。严格来说在 CPython 中多线程因为 GIL 的存在并不能让多个 Python 线程真正并行执行 CPU 计算。GIL 是一个解释器级的互斥锁它保证同一时刻只有一个线程能执行 Python 字节码。但要注意在线程执行到网络请求、文件读写、time.sleep()等待这类操作时GIL 会被释放让其他线程继续工作。所以更准确的说法是threading适合处理“大量时间都花在等待外部资源上的任务”。典型场景包括批量调用 HTTP 接口比如采集公开数据、批量翻译文本、批量调用大模型 API批量下载文件比如图片、日志包、模型权重批量读取和处理本地文件比如把一批 CSV 转成 Excel或者将一批图片做压缩同时监控多个目录、多个端口每个线程负责一个数据源GUI 程序和服务器程序里面用后台线程处理耗时操作避免界面卡死。反过来不适合用threading的场景主要就是 CPU 密集型计算例如图像像素级算法、大数计算、复杂排序循环。这类任务需要真正的多核并行应该用multiprocessing让每个进程单独占有一个解释器或者用ProcessPoolExecutor管理进程池。还需要提醒一点边界问题多线程经常用来批量抓取网页或调用接口但这不等于可以无限并发、无视目标方规则。批量任务开始前必须确认目标网站或服务是否有合法授权、是否有频率限制要求以及是否允许自动化访问。本文所有网络示例只用于本地测试和自有数据的环境生产环境的采集任务要遵守服务条款和当地法规并做好请求频率控制。3. 环境准备装好 Python 就能跑threading是标准库模块环境准备非常简单。你先确认本机 Python 是否可用python --version pip --version如果命令提示找不到python可以换成python3试试这通常意味着 Python 没有加入 PATH或者当前系统只安装了python3命令。如果你是 VS Code 用户安装 Python 扩展后还需要注意右下角解释器是否选中了正确的 Python 环境。为了不在测试阶段污染全局环境建议创建虚拟环境python -m venv venv # Windows venv\Scripts\activate # Linux / macOS source venv/bin/activate本文大部分代码只需要标准库激活环境后直接执行即可。只有网络请求示例会用到第三方库requests可以按需安装pip install requests环境检查参考表检查项预期结果处理建议Python 版本3.8 以上过低版本建议升级python 命令是否可用打印版本号没有就加入 PATH 或使用 python3虚拟环境是否激活命令行前缀出现 venv按对应系统执行 activate 脚本requests 是否安装pip show requests 有输出没有就执行 pip install requests4. 创建线程的几种方式先从最小可运行案例开始。创建一个线程本质上就是告诉threading.Thread一个要执行的函数或可调用对象然后调用start()让它运行调用join()让主线程等待它结束。4.1 用函数作为线程入口import threading import time def work(name): print(f{name} 开始) time.sleep(1) print(f{name} 结束) if __name__ __main__: t1 threading.Thread(targetwork, args(线程-1,)) t1.start() t1.join() print(主线程结束)线程-1 开始 线程-1 结束 主线程结束这里最常见的错误是写成targetwork(线程-1)。如果加了括号work会在主线程中先执行完然后把返回值传给target此时线程内部可能什么都没有执行。正确做法是传函数对象本身参数通过args或kwargs传递。4.2 使用可调用对象除了普通函数任何实现了__call__方法的对象都可以作为线程入口。这种方式适合需要在线程中保存状态的场景。import threading import time class DownloadTask: def __call__(self, url): time.sleep(1) print(fcallable 下载完成: {url}) if __name__ __main__: t threading.Thread(targetDownloadTask(), args(https://example.com/a,)) t.start() t.join()4.3 继承 Thread 并重写 run 方法很多人会在实际项目中看到继承式写法。它的优点是可以用子类属性保存额外数据但缺点是不如函数式写法灵活。新手阶段不建议优先使用import threading import time class MyThread(threading.Thread): def run(self): time.sleep(1) print(子类形式的线程执行完成) t MyThread() t.start() t.join()从实际工程角度看推荐方式仍然是“函数 Thread”或“函数 线程池”。函数式的结构更清晰也更容易做单元测试。5. 线程生命周期控制start、join、daemon线程创建之后不会立刻执行你需要调用start()才会真正进入可运行状态。join()则会让当前线程阻塞等待目标线程结束。没有join()时主线程可能先于子线程结束但程序并不会立刻退出因为 Python 解释器会等待所有非守护线程执行完毕。import threading import time def task(name, delay): print(f{name} 开始) time.sleep(delay) print(f{name} 结束) if __name__ __main__: jobs [ threading.Thread(targettask, args(任务A, 2)), threading.Thread(targettask, args(任务B, 1)), ] for t in jobs: t.start() # 等待所有任务完成 for t in jobs: t.join() print(主线程在所有任务完成后继续)如果没有join()输出中的“主线程继续”会提前出现。写多线程脚本时如果你需要汇总多个线程的结果再继续下一步join()是必须的。5.1 守护线程 daemon 是什么daemonTrue的线程被称为守护线程。守护线程的特点是当主线程结束时它会被强制终止不会阻止程序退出。用一句话记非守护线程是“我要干完再走”守护线程是“主人走了我也走”。import threading import time def worker(): try: print(后台线程开始) time.sleep(3) print(后台线程执行完毕) except Exception as exc: print(f后台线程异常: {exc}) t threading.Thread(targetworker, namedata-fetch-thread, daemonTrue) t.start() print(主线程结束)这里主线程执行结束进程退出后守护线程可能根本没机会执行完time.sleep(3)后面的代码。所以守护线程适合日志上报、定时心跳、临时监听这类“随主线程生死”的任务不适合承载必须落盘的数据处理。在一个线程管理批次任务时如果你不确定“主线程退出后任务是否完整结束”不要轻易设置daemonTrue。5.2 设置线程名称给线程起名字在排查问题时会很有用。默认线程名是Thread-1、Thread-2但并发一多你很难通过默认名定位业务逻辑。创建线程时可以指定name在线程内通过threading.current_thread().name获取import threading def worker(): print(f当前线程{threading.current_thread().name}) t threading.Thread(targetworker, namefetch-stock-price) t.start() t.join()线程常用方法汇总方法/属性作用注意事项start()启动线程一个线程对象只能 start 一次join(timeoutNone)等待线程结束可设置超时避免无限等待daemon标记守护线程主线程退出后守护线程会被强杀name线程名称建议命名为“业务含义-编号”threading.current_thread()获取当前线程对象常用于日志输出threading.enumerate()列出所有存活线程排查线程泄露时使用如果主线程对某个子线程执行join(2)最多等待 2 秒超时后继续向下执行。此时子线程可能仍在后台运行后续如果要使用它的结果必须另行设计等待或取消机制。6. 线程安全与线程锁实战多线程踩得最多、也最容易让人迷惑的坑就是共享变量竞争。先看一段看起来很简单的代码10 个线程每个线程把同一个全局变量加 10 万次直觉上结果应该是 100 万。import threading count 0 def add(): global count for _ in range(100000): count 1 threads [threading.Thread(targetadd) for _ in range(10)] for t in threads: t.start() for t in threads: t.join() print(count , count)多次运行后结果几乎不可能是 100 万而是会比 100 万小。原因在于count 1在 Python 字节码层面并不是“原子操作”它实际上分成多个步骤读取count的当前值计算加 1再把新值写回。线程在执行到中途时可能被切换另一个线程读取到的仍是旧值写回之后就把前一个线程的更新覆盖掉了。解决办法是给共享变量的修改操作加锁。threading.Lock是最基础的互斥锁同一时刻只允许一个线程持有锁并进入临界区。import threading count 0 lock threading.Lock() def add(): global count for _ in range(100000): # with lock 等价于 lock.acquire() lock.release() with lock: count 1 threads [threading.Thread(targetadd) for _ in range(10)] for t in threads: t.start() for t in threads: t.join() print(count , count)这个版本的结果会稳定为 1000000。使用锁时有三个原则值得记住第一锁的范围要尽量小。如果直接把整个for循环包在锁里面那么 10 个线程本质上还是串行执行多线程的并发优势就没了。锁只保护真正会发生竞争的变量修改语句网络的等待、文件的读取、耗时的计算都不应该放在锁内部。第二尽量避免持锁期间再去获取另一把锁。多个线程分别持有不同的锁后互相等待对方释放就会造成死锁。死锁会导致程序卡住不报错也不退出。应对方法是让所有线程按相同的顺序加锁或者优先用RLock。RLock也叫可重入锁它允许同一个线程多次获取锁而不会把自己锁死。第三优先用with lock写法。手动写acquire()和release()时如果中间代码抛了异常release()可能不会执行锁就永远无法释放。用上下文管理器可以避免这个风险。import threading lock threading.Lock() # 推荐with 结束会自动释放锁 with lock: pass # 也可以但必须配合 try/finally否则异常时容易死锁 lock.acquire() try: pass finally: lock.release()7. Queue 与生产者消费者模型实战使用锁可以解决共享变量竞争但更工程化的方式是用queue.Queue做线程间通信。Queue本身是线程安全的多个线程可以安全地往里放数据和取数据不需要你额外加锁。它非常适合搭一个“生产者消费者模型”。下面这个例子模拟一个批量任务处理流程主线程把 20 个任务放进队列4 个worker线程从队列中取任务并处理处理完成后调用task_done()主线程通过q.join()等待所有任务完成。import queue import threading import time def worker(q): while True: item q.get() if item is None: break try: current threading.current_thread().name print(f{current} 处理 {item}) time.sleep(0.2) finally: # 每个任务处理完成后必须调用 task_done q.task_done() if __name__ __main__: q queue.Queue() # 模拟 20 个批量任务 for i in range(20): q.put(f任务-{i 1}) worker_count 4 workers [] for i in range(worker_count): t threading.Thread(targetworker, args(q,), namefworker-{i}) t.start() workers.append(t) # 阻塞直到队列中所有任务都被 task_done q.join() # 停止所有 worker发 None 作为结束信号 for _ in workers: q.put(None) for t in workers: t.join() print(20 个任务处理完成)这段代码里有几个值得强调的细节q.get()会阻塞等待新任务。如果 worker 处理完所有任务后继续循环它会一直阻塞在q.get()所以我们要主动放入None作为“停止信号”。None一定是最后放入的因为只要还有真正的任务没有处理完worker 就可能在取到None之前先取到后续的真正任务。因此q.join()必须在发送停止信号之前执行。q.task_done()必须和q.get()配对。每从队列中取出一个任务处理完成后就要调用一次task_done()否则q.join()会一直阻塞。使用try/finally能保证即使任务处理过程中抛了异常task_done()也会执行主线程不会卡死。生产者消费者模型的优势在于“解耦”。生产者只需要往队列里放任务消费者只需要从队列里取任务两边互不关心对方状态。这种结构天然适合批量任务你有一个几千条数据的列表不需要一次性创建几千个线程只需要固定创建几个worker让队列动态分配任务即可。除了Queue还有一个常用同步工具是threading.Semaphore。信号量可以限制同一时间访问某个资源的线程数量比如限制同时请求外部接口的并发数不能超过 3。虽然线程池也能实现类似效果但信号量更轻量适合“线程数量不能预先确定但必须限制并发”的场景。8. 使用线程池管理批量任务实际开发中手动创建成百上千个线程并不是好方案。线程本身也有创建和销毁成本线程太多反而会导致上下文切换频繁、内存占用增加。更推荐的做法是用concurrent.futures.ThreadPoolExecutor管理线程池。假设有一个包含 20 个任务的列表每个任务模拟耗时 1 秒的网络请求使用 5 个固定线程处理from concurrent.futures import ThreadPoolExecutor, as_completed import threading import time def load_data(name): time.sleep(1) return f{name} - ok({threading.current_thread().name}) names [f任务-{i} for i in range(20)] with ThreadPoolExecutor(max_workers5) as executor: futures [executor.submit(load_data, name) for name in names] for future in as_completed(futures): print(future.result())这段代码的核心流程是executor.submit(load_data, name)会把一个任务提交到线程池立即返回一个Future对象as_completed(futures)会按照任务完成的先后顺序产出结果哪个先完成就先处理哪个with块结束时executor.shutdown(waitTrue)会等待所有任务完成。如果你的业务不要求逐个处理结果而只是希望按原顺序拿最终结果用executor.map()更简单from concurrent.futures import ThreadPoolExecutor def load_data(name): return f{name} processed names [f任务-{i} for i in range(20)] with ThreadPoolExecutor(max_workers5) as executor: results executor.map(load_data, names) for result in results: print(result)在写爬虫、接口调用和下载任务时还需要注意并发数不能盲目调大。线程数过大时真正的瓶颈往往不是本机 CPU而是目标服务端的连接限制或频率限制。常见的起步思路是任务类型建议 max_workers 范围理由批量 HTTP 请求5 - 20需结合目标服务限流情况调整批量下载文件4 - 10既要带宽稳定也要控制连接数本地小文件读写2 - 8磁盘并发太高反而性能下降大模型 API 调用双倍于账号限流值超过限流会频繁报错一个比较稳妥的启动方式是先用max_workers4或max_workers8跑一轮小样本观察耗时和服务端是否报错再逐步调大。不要一上来就用几百个线程去访问外部接口这样做既不稳定也可能违反服务协议。对于生产环境的采集或自动化任务务必先拿到合法授权或者确认使用的是自有服务。9. 性能观察多线程到底快在哪里理解threading适不适用最好的办法是自己跑一遍对比实验。下面用两组代码测量一组是 IO 密集型任务一组是 CPU 密集型任务。9.1 IO 密集型任务对比import threading import time def io_job(seconds): time.sleep(seconds) def run_serial(seconds): start time.perf_counter() for _ in range(8): io_job(seconds) return time.perf_counter() - start def run_thread(seconds): start time.perf_counter() threads [ threading.Thread(targetio_job, args(seconds,)) for _ in range(8) ] for t in threads: t.start() for t in threads: t.join() return time.perf_counter() - start if __name__ __main__: seconds 0.2 serial run_serial(seconds) threaded run_thread(seconds) print(fIO 任务串行耗时: {serial:.3f}s) print(fIO 任务多线程耗时: {threaded:.3f}s)time.sleep模拟的是网络等待、磁盘等待这一类“不占用 CPU 计算”的等待。在常见运行环境下8 个 0.2 秒的任务串行大约是 1.6 秒而多线程版本通常会显著小于 1.6 秒接近其中单个耗时最长的那一个。9.2 CPU 密集型任务对比import threading import time def cpu_job(n): total 0 for i in range(n): total i return total def run_serial(n): start time.perf_counter() for _ in range(8): cpu_job(n) return time.perf_counter() - start def run_thread(n): start time.perf_counter() threads [ threading.Thread(targetcpu_job, args(n,)) for _ in range(8) ] for t in threads: t.start() for t in threads: t.join() return time.perf_counter() - start if __name__ __main__: n 5000000 serial run_serial(n) threaded run_thread(n) print(fCPU 任务串行耗时: {serial:.3f}s) print(fCPU 任务多线程耗时: {threaded:.3f}s)这段计算没有等待外部资源线程始终需要持有解释器执行 Python 字节码。由于 GIL 的限制它无法用多个 CPU 核心并行计算多线程版本通常不会比串行快甚至可能因为线程切换而更慢。这个对比告诉我们判断“要不要用threading”不是看任务数量而是看任务类型。如果一个任务 90% 的时间都在等待网络响应或磁盘读取那么多线程能带来明显收益如果一个任务 90% 的时间都在做运算真正的方案是multiprocessing或者把计算下推到 C 扩展库、NumPy 这类能够释放 GIL 的底层实现。运行实验时可以使用系统自带的任务管理器Windows或htopLinux/macOS观察 CPU 使用率。IO 密集型多线程测试中CPU 使用率不会升高太多因为进程大部分时间在等待CPU 密集型多进程测试中才能看到多个核心被真正利用起来。10. 常见问题与排查方法多线程程序一旦出现诡异问题很多新手会不知道从哪里查。下面整理一份高频问题清单问题现象可能原因排查方式解决方案线程没有执行任何逻辑targetwork()写成了函数调用检查 target 是否传了函数对象改成targetwork参数放args输出结果显示主线程先结束没有调用join()检查子线程后面是否执行了 join在需要汇总结果前调用t.join()共享变量结果小于预期多个线程同时修改变量导致竞争在变量修改处临时加锁测试使用Lock或改用Queue汇总程序卡死没有报错死锁或queue.join()等待未完成在关键位置打印时间点和线程名统一加锁顺序确保每个 get 都 task_done主线程退出后任务中断子线程被设置为 daemon检查线程的 daemon 属性需要完整执行的任务不要设 daemonTrue线程异常后主线程不知道线程内异常默认只在子线程打印查看线程 traceback或加日志在run或任务函数内捕获异常并记录多线程跑 CPU 计算没有提速GIL 导致无法并行执行字节码查看 CPU 使用率是否接近单核CPU 密集改用ProcessPoolExecutor线程数很多但速度没提升创建线程过多或遇到远端限流查看资源占用和服务端响应降低max_workers增加超时和重试join 等待时间过长目标函数内部有死循环或阻塞等待打印线程栈快照设置join(timeout)给目标函数加超时接口批量调用偶尔失败并发过高触发限流或连接池不足检查异常码和连接日志使用信号量限流设置重试策略日志输出顺序混乱多线程同时打印格式化日志并加线程名使用logging模块不要用 print 排查时序排查多线程问题时一个很有效的万能手段是“加日志”。在启动线程前、线程完成时、获取锁前后、queue.put和queue.get前后都打印当前线程名和任务 ID。这样一旦卡住你能立刻看到是哪个线程停在哪里。另一个手段是先降低并发规模把 20 个线程降到 2 个验证业务逻辑是否正确再逐步增加并发。大量并发下出现的问题往往不是业务逻辑本身错了而是共享资源保护没做到位。11. 最佳实践与下一步学习方向把线程创建、锁、队列、线程池综合起来看真正到项目里最推荐的做法是优先用ThreadPoolExecutor线程间通信用Queue需要保护的共享状态用Lock任务日志统一走logging外部接口任务加超时与重试。关于数据竞争一个更彻底的思路是“能不共享就不共享”。比如批量下载任务每个线程只负责下载一个文件并写入独立文件线程之间不需要读写同一个列表那么竞争概率就会大大降低。如果确实需要汇总结果可以让每个任务返回数据最后在主线程里统一收集而不是让多个线程同时修改一个全局 list。对外部请求类任务还应该考虑请求重试、超时、并发限制和输出目录管理。比如下载一批文件文件名最好包含任务标识避免两个线程写到同一个路径。对可能失败的任务建议把失败信息单独记录到一个队列或日志文件中等主流程跑完后统一重试。下面是几个可以直接用起来的工程建议# 批量任务目录规划示例 project/ ├── inputs/ # 待处理任务清单或源文件 ├── outputs/ # 每个线程输出独立文件 └── logs/ └── app.log # 多线程日志统一写入第一次跑多线程任务时先选一个小批次比如 20 条数据确认没有共享变量问题、没有资源竞争、没有接口报错再切换到完整数据集。这样能降低排错成本也能更快看清多线程提速的实际效果。需要特别强调多线程并不是 Python 并发编程的终点。如果你的问题场景是 CPU 密集型计算下一步应该学multiprocessing和concurrent.futures.ProcessPoolExecutor如果你的场景是大量高并发网络连接比如同时维护几千个 WebSocket 连接下一步应该学asyncio如果你的场景是围绕 Redis、数据库连接池或消息队列做任务分发那么今天的Queue模型也会继续帮你理解后续框架的原理。回到今天这一篇最终需要记住的核心就三条第一网络请求和文件等待类任务用多线程有效第二多个线程修改共享变量必须加锁第三批量任务不要自己 open 几百个线程优先用线程池加队列控制并发。把这三点落到代码里threading模块就算是真正掌握了。建议把文中的锁竞争和线程池两个例子保存成模板以后遇到批处理任务时直接改参数复用。