100天精通Python-第37天多线程threading模块基础代码实战如果你搜到这篇文章大概率正被一个问题卡住自己写的 Python 脚本明明能跑但一碰到“批量下载”“并发请求后台接口”“同时处理多个文件”这类任务代码就变成了漫长的等待——一个任务没结束后面的只能排队。于是你打开搜索引擎看到“多线程”“python多线程”“threading模块”这些关键词想给自己的脚本提速又担心线程不好控制、数据会乱、程序会崩。这是一个非常典型的分水岭场景。能写出单线程脚本说明你已经掌握 Python 基础语法但真实项目里的效率问题往往不是“代码写得慢”而是“等待时间没有被利用起来”。多线程就是解决这类等待问题的重要手段。不过这里必须给你一个提前量Python 的线程并不像 Java 或 C 那样天然适合 CPU 密集型并行计算它更擅长解决 IO 密集型任务的并发问题。这篇文章讲清楚两件事第一threading模块的基本用法包括线程创建、启动、等待、守护线程第二从“能跑通”到“用得稳”的细节包括线程安全问题、锁、队列通信、线程池。后面还会给出一套可以直接复制运行的代码实战方便你对照着理解技术点。1. 多线程解决什么问题从一次“等待浪费”说起我们先不念概念直接看一个很多人写过的脚本类型——批量下载文件。假设你要下载 10 个文件每个文件下载需要等待 2 秒。单线程写法很自然import time def download(file_name, wait_time): print(f开始下载 {file_name}) time.sleep(wait_time) print(f下载完成 {file_name}) def main(): start time.time() for index in range(10): download(ffile_{index}.zip, 2) print(f总耗时 {time.time() - start:.2f} 秒) if __name__ __main__: main()这段代码没有任何逻辑错误运行结果也很清晰每个文件下载 2 秒10 个文件总共耗时约 20 秒。问题出在哪里出在“等待”两个字上。下载文件时程序的主要时间花在等待网络数据返回。在这 2 秒里CPU 并没有做多少计算它其实处于空闲或极低负载状态。也就是说单线程执行期间大量时间被“浪费”在等待上。如果改成多线程可以把 10 个下载任务分别交给多个线程去发起请求。当一个线程在等待网络数据时另一个线程可以继续发起新的下载任务。理想情况下如果服务器不限制并发请求数、网络带宽也充足10 个线程同时下载 10 个文件总耗时可能从 20 秒降低到接近 2 秒。这就是线程能带来的价值在等待期间调度执行其他任务提高系统的整体吞吐量。很多初学者容易误解多线程等于“同时做两件事”。这只是宏观印象。准确地说多线程让程序在等待 IO 时不做无谓的空转把原本串行占用的等待时间重叠起来。与其死记“IO 密集型任务适合多线程CPU 密集型任务适合多进程”不如用一个具体的判断标准如果你的程序大部分时间花在网络请求、文件读写、数据库查询、用户输入这类等待操作上就适合多线程如果大部分时间都在做大量数学计算Python 的普通线程帮不上忙原因后面马上会解释。2. 先搞清楚核心原理进程、线程与 GIL在动手写代码之前最好先理解三个基础概念。这些概念也是面试题里反复出现的“多线程基础”搞懂它们后面看代码会轻松很多。2.1 进程与线程的区别进程可以理解为一个正在运行的程序实例。每个进程拥有自己独立的内存空间、文件描述符等资源。一个 Python 脚本运行时操作系统会为它创建一个进程。线程是进程内部的执行单元。同一个进程内的多个线程共享这个进程的内存空间和大部分资源。如果把进程比作一家公司线程就是公司里的员工。公司有独立的办公场地、设备员工共享这些资源但每个人负责不同的任务可以并行推进。线程的一个显著优势是“轻”。创建线程比创建进程快得多线程之间切换的开销也更小。但线程共享内存也带来了问题多个线程同时修改同一个变量时容易出现数据不一致。这就是后文要重点讨论的线程安全问题。2.2 GILPython 线程最容易被误解的点很多 Java 或 C 开发者转到 Python 后会非常疑惑为什么 Python 多线程处理计算密集型任务时运行速度不升反降这里必须提到 GIL也就是全局解释器锁。GIL 是 CPython 解释器的一个设计同一时刻只有一个线程能够执行 Python 字节码。也就是说在单个 Python 解释器进程内多个线程并不能真正利用多核 CPU 同时执行 Python 代码。既然多核并行不行那为什么还要用 Python 多线程关键在于线程遇到阻塞 IO 操作时会主动释放 GIL。比如线程 A 开始下载文件它进入等待状态后会让出解释器锁线程 B 就能获得执行机会。这样在大量 IO 等待的场景里多线程依然可以显著提升效率。再补充一句虽然 Python 线程在 CPU 密集型任务上不如多进程但“因为有 GIL所以 Python 多线程没用”是不准确的说法。GIL 影响的是计算密集型并发不是 IO 密集型并发。很多网络爬虫、接口调用、文件读写脚本都用 threading 获得了倍速提升。2.3 什么时候应该使用多线程根据上面的原理可以列出三个使用建议网络爬虫、批量请求接口、下载文件、发送邮件等网络 IO 密集型任务适合多线程。大量文件读写、数据库批量查询也属于 IO 密集多线程可以提升整体效率。纯 CPU 计算例如图像像素级处理、复杂数学运算、大数组循环不推荐使用 threading应该考虑多进程模块 multiprocessing 或 ProcessPoolExecutor。你可以把“是否有足够多的等待时间”作为第一判断依据。等待时间越多多线程的收益越明显几乎没有等待的计算任务用多进程更合适。3. 环境准备只需要一个 Python 3 解释器这一章内容不依赖第三方库全部基于 Python 标准库所以环境准备非常简单。建议使用 Python 3.7 以上的版本本文示例在 Python 3.10 下验证通过。Windows、macOS、Linux 都可以直接运行。你可以先在自己的电脑上确认 Python 版本python --version如果你的电脑上安装的是python3命令则使用python3 --version确认可以正常输出版本号后创建一个实验目录例如python_thread_example然后在里面新建 Python 文件。文章后续的小节会涉及多个示例建议按下面的文件结构组织代码方便对照运行python_thread_example/ ├── demo_thread_basic.py ├── demo_lock.py ├── demo_queue.py └── demo_pool.py如果后面某个示例运行时报错“模块不存在”需要先检查是否把命令中的threading或concurrent.futures写错。这两个都是标准库正常情况下不需要安装额外包。4. threading.Thread 基础创建线程、启动线程、等待线程先从一个最小示例开始。这段代码会创建两个线程每个线程模拟一个耗时任务然后观察程序的整体耗时。4.1 第一个多线程代码新建文件demo_thread_basic.py内容如下# 文件路径demo_thread_basic.py import threading import time def worker(task_name, wait_time): print(f[{time.strftime(%H:%M:%S)}] 线程 {task_name} 开始) time.sleep(wait_time) print(f[{time.strftime(%H:%M:%S)}] 线程 {task_name} 结束) def main(): start time.time() # 创建两个线程 t1 threading.Thread(targetworker, args(下载任务1, 2), nameT1) t2 threading.Thread(targetworker, args(下载任务2, 2), nameT2) # 启动线程 t1.start() t2.start() # 等待线程执行完毕 t1.join() t2.join() print(f主线程两个子线程都已结束程序耗时 {time.time() - start:.2f} 秒) if __name__ __main__: main()运行python demo_thread_basic.py输出大致如下[14:20:01] 线程 下载任务1 开始 [14:20:01] 线程 下载任务2 开始 [14:20:03] 线程 下载任务1 结束 [14:20:03] 线程 下载任务2 结束 主线程两个子线程都已结束程序耗时 2.00 秒注意观察重点两个线程的“开始”打印几乎同时出现。整个程序耗时约 2 秒而不是 4 秒。这说明两个 2 秒的等待时间被重叠执行了。主线程打印“两个子线程都已结束”时两个子线程已经执行完毕。4.2 start、run、join 的区别Thread(target函数, args参数)只是创建一个线程对象并没有真正开始执行。真正让线程跑起来的方法是start()。这经常是新手犯错的第一个点。有人会误以为调用run()就能启动线程。实际上直接调用run()只会在当前线程中同步执行这个函数完全不产生新线程。你可以把run()理解成“线程对象内部真正要执行的目标函数”而start()才是让操作系统在当前进程里创建一个新的执行流并由这个新执行流去调用run()。来看这个对比示例import threading import time def worker(): time.sleep(1) print(worker 执行完毕) if __name__ __main__: t threading.Thread(targetworker) t.run() # 注意这里不是 start() print(主线程结束总耗时约 1 秒因为是同步执行)运行后你会发现程序先等 1 秒打印 worker 执行完毕然后才打印主线程结束。整个过程没有多线程效果。join()方法用于阻塞当前线程等待被调用线程结束。如果你在主线程中调用t1.join()主线程会在这里暂停直到 t1 执行完毕后继续。如果遗漏 join主线程可能提前执行完自己的代码并退出导致子线程的输出看起来错乱。join()支持超时参数例如t.join(1)表示最多等待 1 秒。如果 1 秒后线程还没有结束主线程会继续向下执行。4.3 守护线程 daemon线程还有一个容易被忽略但非常重要的属性daemon也就是守护线程。默认情况下daemon 属性为 False。这意味着当主线程代码执行完毕时Python 不会立即退出进程它会等待所有非守护线程执行结束。这种机制保证子线程的任务能正常完成。如果把线程设置为 daemonTrue情况则相反主线程执行结束后守护线程会直接被强制终止。示例import threading import time def worker(): while True: print(守护线程工作中...) time.sleep(1) if __name__ __main__: t threading.Thread(targetworker, daemonTrue) t.start() time.sleep(2) print(主线程结束)运行后程序大概只输出两次“守护线程工作中”然后主线程结束进程退出。如果去掉daemonTrue程序会一直运行因为 worker 线程是普通线程它会一直循环不退出。守护线程常用于后台心跳、日志轮转、定时检查等不需要手动收尾的场景。但是需要注意如果在守护线程中执行了重要任务比如写文件到一半、发送消息到一半主线程退出可能造成任务中断。对核心业务任务应该使用非守护线程并配合 join 等待完成或者使用队列和线程池来保证任务被可靠执行。5. 线程安全与锁一个自增程序为什么把结果跑“错”多线程并行执行能让 IO 任务提速但也引入了单线程里不存在的问题多个线程同时修改共享数据会导致数据不一致。下面这个例子非常经典也是很多面试题的原型。5.1 先看一个“跑不对”的例子新建文件demo_lock.py内容如下# 文件路径demo_lock.py import threading counter 0 def increase(): global counter for _ in range(1_000_000): counter 1 def main(): threads [] for _ in range(5): t threading.Thread(targetincrease) threads.append(t) t.start() for t in threads: t.join() print(counter 期望值:, 5_000_000) print(counter 实际值:, counter) if __name__ __main__: main()运行几次你会发现实际值经常不是 500 万而是一个比 500 万小的数。为什么会出现这种情况因为counter 1这行代码看起来只有一步实际上是两条或多条底层指令的组合先读取 counter 当前值然后做加法最后把新值写回变量。假设两个线程同时读取到 counter 的值为 100。线程 A 做了加法得到 101把 101 写回线程 B 在读取时拿到的也是 100它也做了加法得到 101然后写回。最终结果不是 102而是 101。两次加法只生效了一次这就是典型的线程竞争问题。5.2 使用 Lock 解决竞争Python 的threading.Lock可以保证同一时间只有一个线程进入临界区。修改上面的代码# 文件路径demo_lock.py import threading counter 0 lock threading.Lock() def increase(): global counter for _ in range(1_000_000): # 进入临界区前加锁 with lock: counter 1 def main(): threads [] for _ in range(5): t threading.Thread(targetincrease) threads.append(t) t.start() for t in threads: t.join() print(counter 期望值:, 5_000_000) print(counter 实际值:, counter) if __name__ __main__: main()这次运行结果会稳定输出5000000。代码里的with lock:写法要记住。它相当于在进入代码块前调用lock.acquire()在退出代码块后调用lock.release()并且即使代码块内部抛出异常也能保证锁被释放。如果手动调用lock.acquire() counter 1 lock.release()一旦counter 1这中间发生异常release()就不会执行其他线程可能永远拿不到锁程序就会卡死。所以尽量使用with lock:而不是裸的 acquire/release。5.3 加锁会降低性能也不能滥用到这里你可能会有疑问加了锁以后多个线程访问共享变量时不就变成了排队吗这不是又退回串行了吗确实锁的本质就是“牺牲并行度换取数据安全”。在实际工程中要尽可能把锁的范围控制在小范围内只保护真正需要保护的那几行代码而不是在整套任务处理逻辑外层加一把大锁。如果锁范围太大线程在锁保护的临界区内执行很耗时的操作其他线程只能干等多线程的效率优势就会被抵消。更危险的是死锁两个线程各自持有一把锁同时又等待对方释放锁。比如线程 A 持有锁 1等待锁 2线程 B 持有锁 2等待锁 1双方互相等待程序就永久卡住了。避免死锁的常见建议是确保多个线程在获取多个锁时按照相同的顺序去获取或者尽量让锁的粒度小避免嵌套加锁。6. 线程间通信queue.Queue 实现生产者消费者上一节解决的是“多个线程共享同一个变量”的问题。但实际项目中更常见的模式是一个线程负责产生任务多个线程负责处理任务处理完后再把结果收集回来。如果所有线程都直接读写同一个列表变量就需要自己加锁保护还得考虑等待条件非常容易出错。Python 标准库提供了一个更适合这种场景的组件queue.Queue。它是一个线程安全的队列内部已经实现了锁和等待逻辑。多个线程可以同时往队列里 put 数据、从队列里 get 数据而不会出现数据竞争。6.1 生产者消费者代码实战新建demo_queue.py# 文件路径demo_queue.py import queue import random import threading import time TASK_COUNT 8 WORKER_COUNT 3 def worker(worker_id, task_queue): while True: try: task_name, wait_time task_queue.get(timeout0.5) except queue.Empty: # 等待 0.5 秒仍然没有新任务退出循环 break print(f工作线程 {worker_id} 开始处理: {task_name}) time.sleep(wait_time) print(f工作线程 {worker_id} 完成: {task_name}) # 通知队列这个任务已经处理完成 task_queue.task_done() def main(): task_queue queue.Queue() # 生产者向队列中放 8 个任务 for i in range(TASK_COUNT): task_name f任务-{i} wait_time random.uniform(0.5, 1.5) task_queue.put((task_name, wait_time)) # 创建并启动消费者线程 workers [] for i in range(WORKER_COUNT): t threading.Thread( targetworker, args(i, task_queue), daemonTrue ) workers.append(t) t.start() # 阻塞直到队列中的所有任务都被处理完 task_queue.join() print(所有任务已经处理完毕) if __name__ __main__: main()运行python demo_queue.py输出大致如下工作线程 0 开始处理: 任务-0 工作线程 1 开始处理: 任务-1 工作线程 2 开始处理: 任务-2 工作线程 0 完成: 任务-0 工作线程 0 开始处理: 任务-3 ... 所有任务已经处理完毕6.2 这段代码的设计细节第一个关键点是退出条件。消费者线程执行的是while True循环如果队列中已经没有任务线程会一直阻塞在get()上。这里给get设置了timeout0.5线程会等待 0.5 秒仍然没有新任务就抛出queue.Empty异常并退出。如果不加超时也不做退出机制线程会永久阻塞程序不会正确结束。第二个关键点是task_queue.task_done()。这个方法必须由消费者在处理完一项任务后调用。它告诉队列“这个任务已经处理完毕”。主线程里的task_queue.join()会一直阻塞直到所有 put 进队列的任务都通过 task_done 通知完成。第三个关键点是Queue本身是线程安全的。这意味着你不需要再额外加 Lock多个消费者同时 get 同一个任务时不会有两个线程拿到同一个任务。这一点大大降低了并发编程的心智负担。这个模式非常接近真实项目里的“任务队列”架构。框架内部一般会维护一个任务队列主线程往队列里塞任务工作线程从队列里取任务执行。理解了这个模式后面看爬虫框架和任务调度系统的代码都会容易很多。7. 更现代的工程化写法ThreadPoolExecutor 线程池每次手动创建线程对象、调用 start、再调 join代码会随着任务数量增加变得很冗长。而且线程本身也是一种资源创建和销毁都有成本。如果任务有几千个就创建几千个线程系统可能直接崩溃。更稳妥的做法是使用线程池。线程池会在启动时创建固定数量的线程然后长期复用。任务到来时空闲线程取走一个任务开始执行任务执行完毕后线程不会销毁而是回到线程池等待下一个任务。Python 中推荐使用的是concurrent.futures模块中的ThreadPoolExecutor。它封装了线程生命周期代码更简洁也天然支持异常处理和结果获取。7.1 使用 submit 和 as_completed新建demo_pool.py# 文件路径demo_pool.py import time from concurrent.futures import ThreadPoolExecutor, as_completed def download_one(file_id): # 模拟下载耗时 time.sleep(1) return f文件-{file_id} 下载完成 def main(): file_ids list(range(10)) start time.time() # 创建一个包含 4 个线程的线程池 with ThreadPoolExecutor(max_workers4) as executor: # 提交所有任务 future_map { executor.submit(download_one, file_id): file_id for file_id in file_ids } # as_completed 会按任务完成的顺序返回结果 for future in as_completed(future_map): print(future.result()) print(f全部任务执行完毕总耗时 {time.time() - start:.2f} 秒) if __name__ __main__: main()运行python demo_pool.py输出里会依次出现 10 行“文件-X 下载完成”整体耗时比 10 秒短得多。在线程数为 4 的情况下理想耗时接近 3 秒。这里有几个重点with ThreadPoolExecutor(max_workers4) as executor:会在代码块结束时自动调用shutdown()等待所有提交的任务完成并回收线程池资源。executor.submit(fn, arg)会返回一个 Future 对象可以理解为一个“未来结果”的容器。future.result()会阻塞当前线程直到该任务真正完成并返回结果。如果任务内部抛出异常future.result()会把线程内部的异常重新抛出来这样你可以在主线程中捕获。7.2 使用 map 按顺序输出结果如果不需要实时按完成顺序处理结果可以更简单地使用executor.map。它会保持输入顺序对应的输出结果# 文件路径demo_pool.py import time from concurrent.futures import ThreadPoolExecutor def download_one(file_id): time.sleep(1) return f文件-{file_id} 下载完成 def main(): file_ids list(range(10)) start time.time() with ThreadPoolExecutor(max_workers4) as executor: # map 会按照 file_ids 的顺序返回结果 for result in executor.map(download_one, file_ids): print(result) print(f全部任务执行完毕总耗时 {time.time() - start:.2f} 秒) if __name__ __main__: main()运行效果和第一个版本相似但输出顺序是固定的 0 到 9。在爬虫、批量接口调用这类“任务量大、每个任务独立”的业务中推荐优先使用ThreadPoolExecutor。它不需要手动管理线程数组代码整洁程度和维护成本都优于裸写 threading。8. 多线程编程常见问题与排查思路多线程程序最麻烦的一点是很多错误是间歇性的。有时候运行正常有时候结果不对有时候还直接卡死。下面整理一批新手容易遇到的问题按现象、可能原因、排查方式、解决方案四个维度来梳理。问题现象可能原因排查方式解决方案共享变量最终结果不对多个线程同时读写同一变量存在数据竞争检查代码中是否有直接修改全局变量的操作使用 Lock 保护临界区或者使用 queue.Queue 避免直接共享报错 RuntimeError: threads can only be started once对同一个 Thread 对象重复调用 start()检查线程是否在循环中被重复使用每个 Thread 对象只能 start 一次需要新任务时创建新线程对象主线程已经打印结束子线程才输出子线程是非守护线程但主线程没有 join输出顺序看起来混乱观察 print 的顺序在合适的时机调用 join让主线程等待子线程完成后再继续设置了 daemonTrue 后任务执行不完整主线程退出时守护线程被强制终止确认工作线程是否被设置成守护线程对重要任务不要使用 daemonTrue或者在主线程中 join程序卡死没有任何错误输出代码中出现死锁或 get() 一直阻塞没有超时查看是否有嵌套锁或 queue.get() 没有超时参数避免嵌套获取多把锁使用 timeout 超时参数锁代码统一使用 with多线程运行后反而更慢任务本身是 CPU 密集型受 GIL 限制或线程数量过多切换开销大分别统计 CPU 消耗时间和 IO 等待时间计算密集型任务改用多进程线程数不宜超过 CPU 核心数太多print 输出顺序不固定线程调度顺序由操作系统决定不是代码从上到下执行检查是否依赖输出顺序做判断不要依赖线程输出顺序结果顺序通过 Queue 或 Future 去管理如果是加锁后程序仍然偶尔卡死第一步建议把锁的获取和释放全部改成with lock:写法因为手动 release 在异常条件下很容易遗漏。如果使用了多个队列也要注意每个get是否都设置了超时避免因为某个生产者线程崩溃消费者线程永远阻塞等待新任务。还有一点如果程序崩溃不要只盯着业务逻辑可以先在代码入口处加上threading.enumerate()看看到底有哪些线程还在运行。这个打印有助于发现“是不是有线程没退出”的问题。9. 多线程工程使用的最佳实践与后续学习方向看到这里你已经跑通了从基础线程创建到线程池的完整示例。下面这些建议是从“能运行”迈向“能在工程里稳定使用”时需要特别注意的实践要点。第一先用线程池不要手动创建几十上百个线程。手动创建线程本身没有错但线程数量一旦不可控系统资源、调试难度都会迅速膨胀。ThreadPoolExecutor能通过 max_workers 限制并发数同时复用线程这才是生产环境常见的写法。第二考虑任务类型是 IO 密集型还是 CPU 密集型。网络请求、文件读写、数据库等待使用多线程CPU 密集计算建议改成ProcessPoolExecutor或multiprocessing。不要指望 Python 多线程让死循环计算变快这属于用错了工具。第三线程之间尽量少共享可变数据。与其让多个线程共同修改一个全局字典不如把待处理任务放进queue.Queue让每个线程从队列里取任务处理完以后把结果再通过队列或 Future 返回。数据流清晰之后很多并发问题自然消失。第四锁的粒度要小加锁代码统一使用with lock:。如果锁定范围内执行了很耗时的操作其他线程会一直被阻塞。能放在锁外面的操作尽量放在锁外面。第五线程内的异常不能静默吞掉。如果在工作函数里捕获异常后只打印一行日志主线程端不会知道任务失败了。使用 Future.submit 时可以通过future.result()接收异常在线程池外再统一处理失败重试比在每个线程内部处理更可控。第六日志和输出要谨慎。print 在多线程环境下输出顺序不稳定也可能出现内容交错。如果在正式项目里记录执行日志应该使用 Python 的logging模块并开启 logging 的线程安全机制尽量避免多个线程直接 print。第七注意网络 API 的并发限制。很多人写爬虫或调用第三方接口时加了 50 个线程并发请求结果把对方服务器打到请求失败或者自己的 IP 被封。增加并发量之前先看服务方文档是否限制了并发数。如果你打算沿着这个方向继续学下一步可以关注这些内容queue.Queue的内部实现与task_done机制、threading.Event和Condition做更复杂的线程同步、concurrent.futures中的 Future 对象、GIL 内部原理以及多进程模块 multiprocessing。对爬虫场景来说了解线程池与请求库结合时如何控制并发频率也很有实践价值。多线程最值得记住的一句话是它真正的难点不是“启动多个线程”而是并发条件下的数据边界。建议你写这样一个练手项目用线程池批量下载本地的多个文本文件模拟每个文件需要不同耗时再统计总耗时或者实现一个带有任务队列的小型爬虫调度器。只有把代码改成自己的任务你才会真正理解 join 为什么需要调用、task_done 为什么不能漏掉、锁为什么不能随便加在外面。把这些坑都趟过一遍你的 Python 并发基础就扎实了。