简介这份Python课件聚焦第13章“多线程与多进程编程”面向正在学习Python并发编程的初学者与进阶开发者系统讲解多线程在GUI响应、索引服务、软件启动动画等场景中的应用并深入分析多核/单核环境下的线程调度原理及Python GIL对并行性能的限制。课件详细介绍了threading模块的active_count、current_thread、enumerate等常用方法与Thread线程类的使用同时覆盖Event、Condition、Lock、RLock、Semaphore、Timer等同步原语并通过定时器示例和自定义线程类示例演示线程创建、启动与同步的完整流程。资源为单个PPT文件共1个文件容量约717KB便于直接用于教学或自学复习。已有205人浏览学习内容结构清晰重点突出既适合课堂演示讲解也适合课后对照练习是快速掌握Python多线程编程基础知识与实战注意点的实用资料。1. Python 的并发困境先从 GIL 说起在正式碰threading和multiprocessing这两个模块之前先要回答一个问题为什么 Python 的多线程经常被调侃有个屁用答案叫 GIL——全局解释器锁。CPython 解释器在同一时刻只允许一个线程执行字节码多线程在单核上轮流抢锁在标准实现下并不能真正利用多核 CPU 做并行计算。这个设计换来了内存管理的简单和 C 扩展的兼容性代价就是 Python 的多线程天生不适合 CPU 密集型任务。但注意GIL 锁的是执行不是等待。当一个线程发起文件读取、网络请求、数据库查询这类 I/O 操作时它会把 GIL 让出来其他线程趁机执行。于是多线程在 I/O 密集型场景爬虫、下载、Web 服务下依然有实打实的收益。而多进程则是每个进程都有一份独立的解释器和内存空间各跑各的 CPU 核心绕开 GIL 的限制。这一章的教学目标是帮初学者建立这个模型I/O 密集用多线程CPU 密集用多进程两者不能靠感觉选而是由 GIL 和任务性质决定的。后面所有代码都围绕这两个方向展开先掌握一个能判断该用谁的心智模型再动手写代码比背 API 重要得多。2. threading 模块的三种写法与锁机制2.1 线程创建的三种姿势函数、类、线程池Python 的threading模块自 Python 2 时代起就存在API 稳定第三方库依赖最少。最常见的写法是直接传目标函数给Thread构造器。import threading import time def worker(name, delay): for i in range(3): time.sleep(delay) print(f[{name}] 第 {i1} 次执行) if __name__ __main__: t1 threading.Thread(targetworker, args(A, 0.5)) t2 threading.Thread(targetworker, args(B, 0.3)) t1.start() t2.start() t1.join() t2.join() print(所有线程执行完毕)args是传给目标函数的位置参数元组即使只有一个参数也要加逗号这是 Python 元组语法的经典坑。start()之后线程进入就绪状态等待调度join()是阻塞当前主线程直到被调用的线程结束。如果忘记join()主线程可能先跑完子线程的输出会在程序退出时才刷出来造成没跑完的错觉。第二种常见写法是继承Thread类重写run()方法。这种写法适合需要在线程内维护状态的场景比如持有一个重试计数器或连接对象。代码如下import threading class DownloadWorker(threading.Thread): def __init__(self, url, timeout10): super().__init__() self.url url self.timeout timeout self.result None def run(self): # 模拟请求耗时 import time time.sleep(1) self.result f下载完成: {self.url} if __name__ __main__: t DownloadWorker(https://example.com/file.zip) t.start() t.join() print(t.result)注意run()里的代码才是在子线程中执行的而start()内部会自动调用run()。不要在创建线程后直接调用run()——那就成了普通函数调用没有创建新线程。第三种是concurrent.futures.ThreadPoolExecutor它把线程创建、回收和队列管理都封装掉了适合批量提交任务的场景。示例如下from concurrent.futures import ThreadPoolExecutor, as_completed import time def fetch(url): time.sleep(0.5) return fHTTP 200: {url} urls [fhttps://api.example.com/page/{i} for i in range(10)] with ThreadPoolExecutor(max_workers5) as executor: futures [executor.submit(fetch, u) for u in urls] for future in as_completed(futures): print(future.result())max_workers是线程池中同时运行的线程数上限设为 5 意味着 10 个任务排队最多 5 个同时跑。submit()每调用一次就提交一个任务并返回一个Future对象as_completed按完成顺序 yield哪个先完成先处理哪个而不是按提交顺序。这个模型接近现代异步编程的语义比手动管理Thread对象更符合工程习惯。2.2 锁、竞态条件与with语法多线程编程最容易踩的坑是共享变量被多个线程同时读写。下面这个例子两个线程各自对同一个全局变量执行x 1一万次。import threading counter 0 def increment(): global counter for _ in range(10000): counter 1 threads [threading.Thread(targetincrement) for _ in range(2)] for t in threads: t.start() for t in threads: t.join() print(counter) # 结果大概率不是 20000原因在于counter 1在 Python 字节码层面分三步读取 counter、加 1、写回。两个线程可能在读取阶段同时拿到旧值随后各自加 1 再写回其中一个写入被覆盖。这种现象叫竞态条件也是很多面试题问多线程安全吗的根源。标准解法是给临界区加锁。threading.Lock是最基础的三元锁同一时刻只允许一个线程持有import threading counter 0 lock threading.Lock() def safe_increment(): global counter for _ in range(10000): with lock: counter 1 threads [threading.Thread(targetsafe_increment) for _ in range(2)] for t in threads: t.start() for t in threads: t.join() print(counter) # 输出 20000with lock避免了手动acquire()和release()配对不齐的风险——如果代码在中途抛异常with语句也会自动释放锁否则锁被永久占住其他线程全部卡死。这是编程中最容易忽视的 Bug死锁往往不是锁本身的问题而是忘记释放锁。另一种锁是RLock可重入锁。它允许同一个线程多次获取同一个锁而不死锁适合嵌套调用中需要反复进入临界区的场景。如果Lock在同一个线程里连续acquire()两次会直接阻塞死锁RLock内部记录持有者每acquire()一次就把计数加一每释放一次减一只有计数归零才算真正让出锁。2.3 线程间通信Event、Condition 与 Queue线程间光靠锁来同步还不够往往需要互相发信号。threading.Event是最简单的事件通知机制一个线程等待事件另一个线程触发事件。import threading import time start_event threading.Event() def waiter(name): print(f{name} 等待开始信号...) start_event.wait() print(f{name} 收到信号开始干活) def starter(): time.sleep(2) print(发起开始信号) start_event.set() threading.Thread(targetwaiter, args(线程A,)).start() threading.Thread(targetwaiter, args(线程B,)).start() threading.Thread(targetstarter).start()Event.wait()默认阻塞直到set()被调用也可以传超时时间比如wait(5)表示最多等 5 秒。set()之后所有等待该事件的线程同时醒来如果只想唤醒一个线程用Condition的notify()更合适它对应一个资源就绪只通知一个消费者的场景。不过在实际工程里最省心的线程间通信是queue.Queue。它是线程安全的内部自带锁put()和get()不需要再加额外保护。经典的生产者消费者模型一份实现是这样的import threading import queue import time q queue.Queue(maxsize10) def producer(): for i in range(20): q.put(f任务-{i}) time.sleep(0.1) q.put(None) # 终止信号 def consumer(): while True: item q.get() if item is None: q.task_done() break print(f消费者处理: {item}) time.sleep(0.2) q.task_done() threading.Thread(targetproducer).start() threading.Thread(targetconsumer, daemonTrue).start()maxsize10设置队列容量上限put()在队列满时会阻塞天然实现了生产者不能塞爆内存的限流。task_done()配合q.join()可以在主线程知道所有任务是否处理完。这里注意q.put(None)这个哨兵值如果把None当作普通任务放入队列消费者会把它也当成业务数据所以需要在消费者内部显式判断并跳出。3. 多进程核心multiprocessing 的进程池与并发模型3.1 为什么会需要多进程绕开 GIL 的正确姿势多线程在 I/O 密集型场景下表现良好但一旦任务变成纯 CPU 计算比如图像滤镜、矩阵乘法、哈希碰撞等线程调度时争抢 GIL 的开销甚至会把性能拖到比单线程还低。multiprocessing模块正是为此存在的——它通过forkLinux/macOS或spawnWindows创建新进程每个进程有独立的内存空间和独立的 GIL操作系统把这些进程调度到不同 CPU 核上真正并行执行。multiprocessing的设计理念是仿照 threading 的 API但底层换引擎。基本用法如下from multiprocessing import Process import os def heavy_calc(n): result sum(i * i for i in range(n)) print(f进程 {os.getpid()} 计算完成: {result}) if __name__ __main__: procs [] for i in range(4): p Process(targetheavy_calc, args(1000000,)) procs.append(p) p.start() for p in procs: p.join()os.getpid()可以验证不同任务确实跑在不同进程上。这里有一个关键约定multiprocessing的代码必须用if __name__ __main__包裹。因为 Windows 下创建新进程会重新导入主模块如果没有这个保护脚本会无限递归地创建子进程最终抛出异常。即使开发环境是 Linux养成这个习惯也能避免跨平台问题。3.2 进程池Pool.map一行代码实现并行计算逐个手动创建Process对象在任务量很大时并不高效因为每创建和销毁一个进程都有系统调用开销。常见做法是用进程池复用固定数量的 worker 进程。multiprocessing.Pool是传统 APIconcurrent.futures.ProcessPoolExecutor是新式 API两者语法不同底层逻辑一致——都是维护一组常驻进程把任务发进去把结果收回来。from multiprocessing import Pool import time def is_prime(n): if n 2: return False for i in range(2, int(n ** 0.5) 1): if n % i 0: return False return True if __name__ __main__: numbers list(range(1, 100001)) start time.time() with Pool(processes8) as pool: results pool.map(is_prime, numbers) print(f8 进程并行筛选耗时: {time.time() - start:.2f}s) print(f质数个数: {sum(results)})Pool(processes8)指定 worker 进程数经验值是与 CPU 逻辑核心数接近在不清楚机器配置时可以写成Pool()让它默认取os.cpu_count()。pool.map(func, iterable)是整个模块最适合教学的方法它把numbers列表切成多块分发给各进程收集所有返回值并组装成一个列表顺序与输入一致。比apply_async的手动收集result.get()要直观得多。另外把is_prime换成 I/O 操作再跑一遍会发现并发加索引的效果明显更差因为进程间的通信开销和上下文切换远大于线程调度。这是判断任务类型的重要信号不区分任务类型就套用多进程性能可能反降。3.3 数据共享的问题为什么不能直接改全局变量进程与线程最大的差异在于线程之间是共享地址空间的全局变量天然可见进程之间拥有完全独立的内存子进程对全局变量的修改不会同步到父进程——因为fork出来的时候子进程持有的是父进程内存的副本。from multiprocessing import Process shared_list [] def append_item(item): shared_list.append(item) print(f子进程看到: {shared_list}) if __name__ __main__: p Process(targetappend_item, args(1,)) p.start() p.join() print(f主进程看到: {shared_list}) # 依然是删除线空列表子进程打印的结果在父进程中完全不存在。这是很多初学者从threading过渡到multiprocessing后踩的第一个坑。解决方案有两条路一是用multiprocessing.Value和Array做底层共享内存二是用Queue传数据让拥有权转移而非共享。下面是一个用Queue实现多进程累加的等价写法from multiprocessing import Process, Queue def producer(q, n): total 0 for i in range(n): total i q.put(total) if __name__ __main__: q Queue() procs [Process(targetproducer, args(q, 1000000 * i)) for i in range(1, 5)] for p in procs: p.start() final_total 0 for _ in procs: final_total q.get() for p in procs: p.join() print(f最终结果: {final_total})Queue实现了进程间安全传递对象的核心机制——pickle序列化。父进程把数据序列化后通过管道或共享内存发给子进程子进程反序列化拿到的是一个新的对象不再共享内存。好处是不需要锁来保护坏处是数据量大时序列化开销不可忽略所以进程间通信频率高、数据量大的任务要重新考虑架构。4. 选择判断I/O 密集与 CPU 密集的试算方法与观察指标4.1 一个能说服自己的实验对比与其背结论不如亲手做一个对照实验。同样跑 10 个 CPU 密集型任务分别用单线程、多线程和多进程实现对比耗时。import time import threading from multiprocessing import Pool def busy_work(n): total 0 for i in range(n * 100000): total i ** 2 return total def run_thread(): threads [threading.Thread(targetbusy_work, args(100,)) for _ in range(10)] for t in threads: t.start() for t in threads: t.join() def run_process(): with Pool(processes10) as pool: pool.map(busy_work, [100] * 10) if __name__ __main__: start time.time() run_thread() print(f多线程耗时: {time.time() - start:.2f}s) start time.time() run_process() print(f多进程耗时: {time.time() - start:.2f}s)注释里写明了两者的预期单线程是基准线多线程通常略慢甚至持平多进程接近单线程分为核心数份的理想耗时。建议在至少 4 核的机器上跑效果会更明显。如果要更精确地观察进程数对性能的影响可以在Pool(processes...)中传入 2、4、8、16 分别测试看耗时曲线在哪里趋于平缓——一般阈值是 CPU 核心数。4.2 该看哪些指标CPU 占用率与阻塞时间如果用眼睛感受不到差距拿数据说话。多线程的 CPU 密集型任务在htop或任务管理器里会看到整体 CPU 占用率不到 100%——因为 GIL 让开了空闲等待实际上只有一个核心在跑。多进程则能看到所有核心被占满。更细致的判断方法是看单一任务中阻塞时间占比。如果任务里sleep、recv、read这类阻塞调用占总执行时间超过一半它就是 I/O 密集型用多线程如果for循环、数学运算占比超过一半优先多进程。一个更经验的判断角度是从工程角度I/O 密集用多线程省内存CPU 密集用多进程占内核——前者开几千个线程内存都兜得住后者每个进程都有一份解释器副本内存开销是按量级翻倍的。4.3 一个经常被忽略的第三方角度协程的存在在 Python 3.4 之后的asyncio提供了一种单线程协作式并发它用async/await语法把一次性阻塞改成挂起-恢复键盘上的线程池和进程池都被比下去。协程的优点是没有线程切换开销缺点是不能真正并行不能并发做 CPU 密集。实际常见的是把asyncio用来做大量网络请求再用multiprocessing做少量重计算服务。需要了解一个背景在 Python 3.13 及以后存在 no-GIL 的实验性构建但绝大多数第三方 C 扩展仍未适配生产环境依然默认 GIL 模式。5. 面向实践的线程池与进程池5.1 用 ProcessPoolExecutor 实现 CPU 密集型效果concurrent.futures.ProcessPoolExecutor是multiprocessing.Pool的高级版它提供与ThreadPoolExecutor一致的submit/map/as_completed接口切换串行并发时只需改导入和名称。from concurrent.futures import ProcessPoolExecutor, as_completed def hash_password_str(password): import hashlib return hashlib.sha256(password.encode()).hexdigest() passwords [secret123, admin, qwerty, letmein] * 25 with ProcessPoolExecutor(max_workers8) as executor: futures {executor.submit(hash_password_str, pwd): pwd for pwd in passwords} for future in as_completed(futures): original futures[future] print(f{original} - {future.result()})max_workers的默认值是 CPU 核心数但实际选择的考虑因素还包括内存限制。比如 8 核机器同时跑 8 个进程每个进程用 300MB那 8 个就占了 2.4GB明显超过一台小内存机器的承受能力。此类问题常见的做法是用getconf _NPROCESSORS_ONLN查逻辑核心数再乘一个小于 1 的系数给系统留出余量。超频、虚拟机和容器环境下的核心数不等同于可用线程数一定要结合cgroup限制来判断。5.2 用 ThreadPoolExecutor 批量下载的完整范例线程池处理 I/O 密集型任务的核心思路是高并发下保持资源上限可控。下面是一个通用下载函数负责批量拉取 URL 并写入文件中途不依赖第三方库适合作为教学示例import urllib.request import os from concurrent.futures import ThreadPoolExecutor, as_completed def download_file(url, savedirdownloads): filename url.split(/)[-1] or index.html filepath os.path.join(savedir, filename) urllib.request.urlretrieve(url, filepath) return filepath if __name__ __main__: os.makedirs(downloads, exist_okTrue) urls [ fhttps://example.com/files/{i}.zip for i in range(20) ] with ThreadPoolExecutor(max_workers8) as executor: future_to_url {executor.submit(download_file, url): url for url in urls} for future in as_completed(future_to_url): print(f完成: {future.result()})urllib.request.urlretrieve在下载文件时会发生阻塞 I/O这正是多线程发挥优势的场景。需要注意filename的提取在当前示例中假设了 URL 一定含有文件名真实场景下要靠 Content-Disposition 头或解析后的路径来控制。future_to_url这层映射能让你在拿到结果时知道它对应哪个 URL便于错误处理和日志记录。如果某个 URL 连不上直接把urlretrieve包进try/except捕获URLError再把 future 的异常绑定到该 URL 上返回一个失败列表而不是让整个程序中断。5.3 线程池在 Web 场景的代表示例给 Web 服务加线程池最常见的是gunicorn的 worker 配置。gunicorn默认使用syncworker即一个 worker 一个进程只处理一个请求如果代码里阻塞在数据库查询上新请求只能排队。改成threads类型的 worker 后每个进程内部可以开多个线程处理并发请求对 I/O 密集型服务有明显提升gunicorn -w 4 --threads 8 app:app这条命令启动 4 个进程每个进程 8 个线程总共支持 32 个并发请求同时处理。严格讲这里的线程池由 gunicorn 内部管理使用--threads 8后无需再手动创建ThreadPoolExecutor。如果应用本身用到了任务队列如 Celery那么 worker 并发模型与线程池是两回事不要混淆。需要强调的是max_workers不是越大越好。线程太多导致 GIL 切换频繁同步开销抵掉 I/O 收益进程太多导致内存占用和进程切换成本变高。经验上I/O 密集型线程数可以先取CPU 核心数 * 10起跑CPU 密集型进程数取CPU 核心数或核心数 1再通过压测试探瓶颈。5.4 调试工具与定位死锁的技巧多线程死锁很难复现常见表现是程序卡住但 CPU 占用为零。相对实用的定位方法是程序运行时发送SIGABRT让 Python 打印所有线程的栈跟踪信息kill -ABRT pidPython 收到该信号后会在 stderr 输出各个线程的栈帧通过最后等待的位置能判断是否卡在lock.acquire()上。faulthandler模块更规范地做这件事——在代码开头启用运行时按CtrlC即打印线程栈import faulthandler faulthandler.enable()对于multiprocessing死锁常见情况是Queue的get()阻塞了而 producer 还在等待join()这种对称等待需要格外小心。排查时先看进程状态ps -ef | grep python查看各进程有没有 CPU 占用如果全部趋近于 0基本就是活锁或死锁再检查队列大小和put/get的配对。6. 在 asyncio 与多进程之间选型并发方案的最后一块拼图6.1 协程与线程的本质差异谁在让路前五章基本覆盖了多线程与多进程但现代 Python 工程里还横着一个更轻量的并发方案——协程。常见做法是在 I/O 密集型任务里优先考虑asyncio因为它的上下文切换是协作式而不是抢占式的由await显式让出控制权省掉了操作系统的线程调度开销。多线程的让路是被迫的——GIL 在预定的时间片或字节码间隔后打断线程执行协程的让路是自愿的——await告诉你解释器我要等待 I/O你先干别的。这导致协程的并发上限比线程高得多因为一个线程的栈内存动辄 8MB而协程只占几 KB。大规模连接场景服务器同时挂着上万条 WebSocket 连接用线程池会耗尽内存用协程则轻松得多。import asyncio async def fetch_page(url): print(f开始请求: {url}) await asyncio.sleep(1) return f完成: {url} async def main(): urls [fhttps://example.com/page/{i} for i in range(10)] tasks [asyncio.create_task(fetch_page(url)) for url in urls] results await asyncio.gather(*tasks) for res in results: print(res) if __name__ __main__: asyncio.run(main())asyncio.gather一次性等待所有任务完成并按任务顺序返回结果。asyncio.sleep模拟真实 I/O 阻塞期间事件循环继续调度其他任务。注意main()本身必须是async def函数因为await只能在协程中使用——这种传染性是初学 asyncio 最大的障碍所有调用链都要改成异步的。6.2 选择依据单层并发还是两层并发实际项目里任务往往不是纯粹的单一类型。比如一个爬虫系统外层要从几十万个 URL 中下载页面是 I/O内层解析 HTML 时要做正则匹配、字符串处理即使不是重计算也涉及 CPU 消耗。这种混合负载没有银弹但常见的架构是协程负责海量 I/O线程池负责少量阻塞型调用如某些不支持 async 的 SDK多进程负责真正吃 CPU 的计算服务。import asyncio from concurrent.futures import ThreadPoolExecutor executor ThreadPoolExecutor(max_workers4) def blocking_db_query(user_id): import time time.sleep(0.5) return fuser_{user_id}_data async def handler(user_id): loop asyncio.get_running_loop() result await loop.run_in_executor(executor, blocking_db_query, user_id) print(result) async def main(): await asyncio.gather(*(handler(i) for i in range(10))) if __name__ __main__: asyncio.run(main())loop.run_in_executor是桥接同步代码和异步事件循环的关键函数它把阻塞函数丢进线程池执行返回一个可await的对象事件循环在主线程等待期间还能干别的事。这种写法避免了在协程里直接调同步阻塞函数导致整个事件循环卡死的低级错误是从 asyncio 入门到实际工程绕不开的第一步。6.3 启动参数与行为验证无论选哪种方案运行时的行为都要能观测。一个很快验证同步阻塞是否卡住协程的实验在协程里直接调用time.sleep(2)而不await观察多个请求是否串行改成await asyncio.sleep(2)后再看耗时能明显看到并发效果。这组对比适合作为代码上线前的快速质量检查。如果要在同一个服务里同时跑多进程和 asyncio常见组合是主进程运行事件循环每个 worker 通过multiprocessing启动一个独立进程。此时需要避免在子进程内再创建事件循环否则会出现 socket 描述符跨进程的状态混乱。架构上尽量让每个模块只选一种并发模型跨模块用Queue做数据传递这是工程上维护性最高的方式。至于多进程与 GPU 计算、C 扩展等场景注意这些资源本身有自己的锁与上下文多进程是否能有收益必须经过压测验证不能只看核心数做结论。并发方案没有唯一正解threading、multiprocessing、asyncio各自代表了轻量并发并行计算高吞吐 I/O三个方向能在同一个项目中准确区分它们的使用边界才是 Python 并发编程真正需要掌握的技能。本文还有配套的精品资源点击获取