简介多线程与多进程是Python并发编程的重要基础。这份PPT课件面向初中级学习者系统讲解多线程与多进程的核心概念涵盖界面响应、后台索引、软件启动动画等典型应用场景以及多核平台线程调度、单核CPU下的适用性并重点分析全局解释器锁GIL对并行性能的限制。课件梳理了threading模块常用方法、Thread线程类的两种创建方式并详细介绍Event、Condition、Lock、RLock、Semaphore、Timer等同步原语的用途与写法配有可运行的定时器示例和自定义线程类代码便于教师直接用于课堂演示也适合学生课后对照练习。资源为单个PPT文件容量717KB结构紧凑、章节清晰适合作为Python课程第13章的教学课件或系统复习资料。目前已有205人学习可帮助初学者快速掌握线程管理、同步机制与多进程编程思路夯实并发编程基础。1. 多线程到底解决了什么问题从 GUI 卡顿到 GIL 前提先抛一个反直觉的结论在单核单 CPU 的机器上Python 多线程不仅不会让程序跑得更快反而可能因为线程切换开销而更慢。但这不妨碍多线程成为 GUI 应用、网络服务、文件索引等场景的刚需——因为它解决的核心问题从来不是“算得更快”而是“别让用户干等”。一个典型的例子是字处理软件用高优先级线程接收键盘输入用低优先级线程做拼写检查和分页统计主线程才不会因为后台任务而阻塞界面刷新。再比如下载器里的断点续传本质上是把一个大文件切成多个区间每个线程负责一个区间的读写配合 HTTP 的 Range 头实现分片下载速度提升的直接来源是并行 I/O 而不是并行计算。这正是本课件第 13 章要讲清楚的核心问题什么时候该用多线程什么时候该用多进程以及 Python 的 GIL 在这中间扮演了什么角色。这篇文章把这套东西拆开讲透适合正在学 Python 并发编程、准备面试、或者要在实际项目里选型的人读。2. threading 模块的顶层 API先摸清现场再动手2.1 为什么要先掌握模块级函数课件 13.1 节列出的active_count()、current_thread()、enumerate()、stack_size()这些函数看起来不起眼但在排查线程泄漏、确认线程栈配置、判断当前代码运行在哪个线程时它们是第一手工具。很多人一上来就写Thread(target...)出了问题却不知道怎么看现场最后只能靠 print 硬猜。实际上Python 的 threading 模块提供了一组内置的“观测点”用好了能省大量调试时间。 import threading threading.stack_size() 0 threading.stack_size(64 * 1024) 0 threading.stack_size() 65536先看stack_size()不传参数时返回当前线程栈大小0 表示使用系统默认值传入参数则设置后续创建的线程的栈大小必须是 0 或大于 32K 的正整数。这个参数在递归深度很大的场景比如深度遍历目录树里会影响RecursionError出现的时机所以监控线程栈大小不是一个没有意义的操作。再看active_count()和enumerate()它们能告诉你当前进程里有多少存活的线程以及它们各自是什么对象这是诊断“线程为什么没退干净”的起点。import threading def worker(): pass t threading.Thread(targetworker, nameworker-1) t.start() print(threading.active_count()) # 存活线程数量包含主线程 print(threading.current_thread()) # 当前执行这段代码的线程对象 print(threading.enumerate()) # 所有存活的 Thread 对象列表 print(threading.get_ident()) # 当前线程的标识符 print(threading.main_thread()) # 主线程对象即启动解释器的线程current_thread()和get_ident()的区别要分清前者返回Thread实例能拿到name、daemon等属性后者只返回一个非负整数标识符无实际语义且可能被复用。调试日志里推荐两个都打因为线程可能被重命名但 ident 在存活期间是稳定的。下面是课件中这些函数的完整能力一览。函数返回值典型用途active_count()int判断是否存在线程泄漏配合日志监控current_thread()Thread 对象在公共函数里判断调用者身份get_ident()int作为 dict 的 key 保存线程私有数据enumerate()Thread 对象列表遍历所有存活线程做清理或诊断main_thread()Thread 对象确认当前线程是否为主线程stack_size([size])int获取或设置线程栈大小2.2 Timer 类的正确用法和 cancel 的边界课件 13.1 里的Timer示例是很多人忽略的一个工具。它本质上是一个“延迟执行”的线程创建时不立即运行调用start()后等待指定秒数再调用目标函数。实现定时心跳、延迟重试、超时清理这类任务时比手动 sleep 后调函数要干净得多。import threading def demo(v): print(fdelay called: {v}) t threading.Timer(3, demo, args(5,)) t.start() print(timer started, waiting...) # 如果还想取消 # t.cancel()Timer(interval, function, argsNone, kwargsNone)的四个参数分别是延迟秒数、目标函数、位置参数、关键字参数。注意一个关键点cancel()只能在定时器还在等待阶段时生效一旦回调已经进入执行cancel()就无能为力了。源码里 cancel 做的是清除底层 Event 并标记_tstate.lock所以它无法中断一个已经开始执行的函数。这个边界在实际项目里很重要比如做超时重试时不能依赖cancel()来中断一个卡死的 I/O 调用而是要去设置一个标志位让被调函数自己判断是否应该退出。3. Thread 对象实战start、join、daemon 的配合逻辑3.1 两种创建线程的方式课件 13.2 给出了两种创建线程的方法向Thread构造函数传入可调用对象或者继承Thread类重写run()。从工程角度后者更适合“线程要携带状态”的场景。比如一个下载任务需要记录进度、错误次数、目标 URL把这些字段放进自定义线程类里代码组织比把一堆参数塞进args清晰得多。import threading class DownloadThread(threading.Thread): def __init__(self, url, save_path): threading.Thread.__init__(self) self.url url self.save_path save_path self.progress 0 def run(self): # 模拟下载每 0.1 秒更新一次进度 for i in range(10): self.progress (i 1) * 10 # 实际项目中这里会写文件、请求网络 print(fdone: {self.url} - {self.save_path}) t DownloadThread(https://example.com/a.zip, /tmp/a.zip) t.start() t.join() print(t.progress)这段代码里最值得注意的细节是自定义线程类的__init__里必须显式调用threading.Thread.__init__(self)否则底层线程状态没有初始化start()会抛RuntimeError: thread.__init__() not called。run()方法在start()调用后自动执行线程结束的标志就是run()返回。关于start()和run()的区分一个常见的初学者错误是直接调用run()那样线程代码确实会执行但执行环境是当前线程而不是新线程完全失去了并发的意义。3.2 join 的本质是等待而非终止课件例 13-1 里的join(timeout)参数很容易被误解。join()的作用是阻塞调用它的线程直到被调线程结束或超过 timeout 秒数。它不是一个终止线程的方法而是“等待”的同步原语。看一个更接近实际的例子import threading import time def worker(name, delay): for i in range(3): print(f{name}: {i}) time.sleep(delay) t1 threading.Thread(targetworker, args(A, 0.5)) t2 threading.Thread(targetworker, args(B, 0.1)) t1.start() t2.start() t1.join(2) # 等待 t1 最多 2 秒 print(t1 alive:, t1.is_alive()) # 若线程未结束返回 True t2.join() # 不限时等待 t2 print(t2 alive:, t2.is_alive())join(timeout)返回后线程可能还在运行is_alive()是检验这一点的标准方法。课件例 13-2 里注释掉join()的两行代码后结果会变化原因就在于此不调用join()主线程继续往下走打印is_alive()时子线程可能还没来得及结束。这提示了一个重要结论需要确保子线程完成后再做后续操作就必须join()不join()也能等其他代码执行完后由解释器回收但那是不确定的行为。3.3 daemon 属性的行为差异是调试陷阱的重灾区课件例 13-3 演示的 daemon 属性是面试常考、也是实际踩坑最多的点。规则本身不难daemon 为 False 的子线程主线程退出时会等待它结束daemon 为 True 时主线程直接退出守护线程被强制终止。但课件里特意提到了一个细节——这个规则在 IDLE 里和 cmd 里表现不一样。原因在于 IDLE 的主线程并不是 Python 解释器进程的入口线程它有自己的事件循环所以 daemon 子线程的行为会受到 IDLE 自身调度的影响。这给我们的实战启示是验证 daemon 行为时应该用python script.py在终端里跑而不是在 IDE 的交互环境里观察。import threading import time class mythread(threading.Thread): def __init__(self, num, name): threading.Thread.__init__(self, namename) self.num num def run(self): time.sleep(self.num) print(self.num) t1 mythread(1, t1) t2 mythread(5, t2) t2.daemon True t1.start() t2.start() # 主线程执行到这里t1 是 False会等 1 秒t2 是 True不会等 print(main exiting)这段代码在 cmd 里运行输出顺序是先打印 main exiting1 秒后打印 1然后进程退出t2 的 5 秒后打印永远不会出现。这里的参数要点是daemon属性必须在start()之前设置否则会抛RuntimeError: cannot set daemon status of active thread。设计上守护线程适合做心跳、日志刷盘、监控上报这类“随进程生死”的任务而数据落地型任务必须是非守护线程否则进程退出时数据可能没写完。4. 同步原语选型Lock、RLock、Event、Condition、Semaphore 怎么用4.1 为什么需要锁从“分片下载的进度累计”说起多线程共享同一份数据时如果没有同步机制就会出现竞态条件。拿分片下载举例——每个线程下载一个分片下载完成后要把进度累加到一个共享变量里。这个count count 1在字节码层面不是原子的两个线程可能同时读到同一个旧值导致进度丢失。Lock就是用来给这段临界区加互斥的。import threading progress 0 lock threading.Lock() def update_progress(): global progress for _ in range(100000): with lock: progress 1 threads [threading.Thread(targetupdate_progress) for _ in range(4)] for t in threads: t.start() for t in threads: t.join() print(progress) # 期望 400000无锁时会出现明显偏差with lock:语句比lock.acquire()/lock.release()手动配对更安全因为即使临界区里抛异常锁也会在退出with块时自动释放。需要注意Lock是不可重入的同一线程连续acquire()两次会死锁。这就是为什么嵌套同步的代码里要用RLock——它允许持有锁的线程再次获取同一把锁内部用计数器记录重入次数每次acquire()必须有对应的release()。4.2 Event一个标志位搞定线程间的“通知”Event的核心是内部的一个布尔标志set()把它置 Trueclear()置 Falsewait(timeout)阻塞直到标志变为 True 或超时。这个机制非常适合做“优雅停机”信号。比如后台轮询线程主线程想让它停止时set()一个事件轮询线程在每个循环里检查这个标志决定是否退出。这比直接用threading.Thread的_stop私有方法可靠得多。import threading import time stop_event threading.Event() def loop_worker(): while not stop_event.is_set(): print(polling...) time.sleep(1) print(worker stopped) t threading.Thread(targetloop_worker) t.start() time.sleep(3) stop_event.set() # 通知线程停止 t.join() print(main exit)这里的关键是wait(timeout)和is_set()的区别wait(timeout)用于“阻塞等待直到事件发生或超时”适合做主线程等待子线程完成某个阶段is_set()用于非阻塞检查适合轮询循环里判断条件。使用场景上一个常见的误区是clear()之后又有新事件到达可能导致通知丢失所以Event适合一次性的通知不适合“事件计数”类的需求——那种需求应该用Semaphore或Queue。4.3 Condition生产者消费者模型的正确姿势Condition比Event更进一步它允许线程等待一个“条件”满足后再继续并且可以做到精确唤醒。课件里列出的场景是消费者线程等待队列非空生产者线程放入数据后通过notify()唤醒一个等待的消费者。看一个教科书级实现import threading import time queue [] condition threading.Condition() def producer(): for i in range(5): with condition: queue.append(i) print(fproduce {i}) condition.notify() # 唤醒一个消费者 time.sleep(0.5) def consumer(): while True: with condition: while not queue: condition.wait() # 自动释放锁并阻塞被唤醒后重新获取锁 item queue.pop(0) print(fconsume {item}) if item 4: break t1 threading.Thread(targetproducer) t2 threading.Thread(targetconsumer) t1.start() t2.start() t1.join() t2.join()代码里最容易被忽略的是while not queue:而不是if not queue:——这是防御虚假唤醒的标准做法。即便condition.wait()被唤醒也要重新检查条件因为可能另一个线程抢先消费了队列里的数据。wait()的底层行为是释放锁阻塞当前线程被notify()或notify_all()唤醒后重新尝试获取锁然后返回。因此wait()必须在with condition:块内调用否则抛RuntimeError。生产环境里如果多消费者争抢优先用notify_all()否则容易出现“唤醒一个消费者但该消费者消费不了其他消费者又没被唤醒”的活锁。4.4 Semaphore限制并发数的最轻量方案Semaphore维护一个计数器acquire()使计数器减一计数器为零时阻塞release()使计数器加一并唤醒一个等待者。最常见的用法是控制同时访问资源比如数据库连接池、API 接口的线程数量。import threading import time sema threading.Semaphore(3) # 最多允许 3 个线程同时进入 def api_call(idx): with sema: print(frequest {idx} start) time.sleep(1) print(frequest {idx} done) threads [threading.Thread(targetapi_call, args(i,)) for i in range(6)] for t in threads: t.start() for t in threads: t.join()Semaphore(3)初始化后计数器为 3前 3 个线程的acquire()立即返回第 4 个线程阻塞直至某个线程release()。注意with sema:语法糖保证release()一定会执行即使time.sleep(1)抛异常。一个常见误用是用Semaphore做互斥锁——计数初始化为 1 时它确实等价于Lock但Semaphore不可重入而RLock可重入所以单纯的互斥场景选RLock更稳妥。在 Python 3.9 及以上版本还有个BoundedSemaphore它在release()次数超过初始值时抛ValueError能及时暴露多余的release()调用代码审查时更容易发现问题。5. 多进程与 GILProcessPoolExecutor 的正确打开方式5.1 GIL 不是洪水猛兽但它决定了你该怎么选GIL全局解释器锁是 CPython 解释器的一个机制同一时刻一个进程里只有一个线程能执行 Python 字节码。所以 CPU 密集型任务用多线程在 Python 里是达不到并行效果的——4 个核只会在一个核上做时间片轮转甚至因为切换开销变得更慢。但 I/O 密集型任务不一样网络请求、文件读写、数据库查询在等待时会让出 GIL其他线程才有机会执行。因此并发场景选多线程并行计算选多进程这是 Python 并发编程里最核心的判断题。课件第 13 章对 GIL 的定位也是如此——它解释了为什么 Python 引入了多进程模块。5.2 用 ProcessPoolExecutor 替代手写 multiprocessingmultiprocessing模块的ProcessPoolExecutor是 Python 3.2 提供的进程池高层接口处理 CPU 密集型任务时比手写ProcessQueue省事得多异常处理和结果获取也更规范。from concurrent.futures import ProcessPoolExecutor import math def calc_factorial(n): result 1 for i in range(2, n 1): result * i return result if __name__ __main__: nums [100000, 100001, 100002, 100003] with ProcessPoolExecutor(max_workers4) as executor: futures [executor.submit(calc_factorial, n) for n in nums] for future in futures: print(future.result())要点一if __name__ __main__是必须的否则 Windows 下multiprocessing会递归导入模块并创建无限子进程。要点二max_workers设置为 CPU 核数时并行效率最高这可以用os.cpu_count()获取。要点三提交任务是submit()返回一个future对象调用.result()会阻塞等待计算结果批量提交还可以用executor.map()它返回迭代器结果按传入顺序依次产出——如果某个任务是死循环map()会卡在第一个结果上而submit()可以配合as_completed()按完成顺序处理。5.3 进程间数据传递的坑pickle 是绕不开的边界多进程之间默认通过 pickle 序列化传递数据这对参数和返回值有硬性限制lambda 函数、局部内部函数、某些动态对象没法 pickle强行传递会抛PicklingError。线程可以共享内存进程则需要通过multiprocessing.Queue、Pipe或共享内存来实现。记住这个对照规则小数据用返回值传递大数据用Queue超大数组用SharedMemory。下面的例子展示了进程池 Queue的完整协作方式from multiprocessing import Process, Queue def worker(q, idx): q.put(idx * 2) if __name__ __main__: q Queue() procs [Process(targetworker, args(q, i)) for i in range(4)] for p in procs: p.start() for p in procs: p.join() while not q.empty(): print(q.get())Queue在进程间传递数据时会自动做序列化和加锁所以它是进程间数据共享的最安全选择。但要留意q.empty()不是 100% 可靠的——另一个进程可能正在 put 数据的半途中此时判断空会误判。更稳妥的做法是先join()所有进程确认它们都已完成写入再逐条get()并用一个哨兵值标记队列结束而不是依赖empty()。最后再强调一次ProcessPoolExecutor它把进程的创建、销毁、结果回调都由框架管理日常 90% 的 CPU 密集型并行需求用它就够了只有当你需要进程间长期通信、共享状态或精细控制子进程生命周期时才值得直接使用multiprocessing.Process和Queue。本文还有配套的精品资源点击获取