企业官网建设流程全解析

简介：这份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}] 第 {i+1} 次执行") if __name__ == "__main__": t1 = threading.Thread(target=worker, args=("A", 0.5)) t2 = threading.Thread(target=worker, 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, timeout=10): 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 f"HTTP 200: {url}" urls = [f"https://api.example.com/page/{i}" for i in range(10)] with ThreadPoolExecutor(max_workers=5) 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(target=increment) 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(target=safe_increment) for _ in range(2)] for t in threads: t.start() for t in threads: t.join() print(counter) # 输出 20000

with 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(target=waiter, args=("线程A",)).start() threading.Thread(target=waiter, args=("线程B",)).start() threading.Thread(target=starter).start()

Event.wait()默认阻塞直到set()被调用，也可以传超时时间，比如wait(5)表示最多等 5 秒。set()之后所有等待该事件的线程同时醒来，如果只想唤醒一个线程，用Condition的notify()更合适，它对应"一个资源就绪，只通知一个消费者"的场景。

不过在实际工程里，最省心的线程间通信是queue.Queue。它是线程安全的，内部自带锁，put()和get()不需要再加额外保护。经典的生产者消费者模型，一份实现是这样的：

import threading import queue import time q = queue.Queue(maxsize=10) 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(target=producer).start() threading.Thread(target=consumer, daemon=True).start()

maxsize=10设置队列容量上限，put()在队列满时会阻塞，天然实现了"生产者不能塞爆内存"的限流。task_done()配合q.join()可以在主线程知道所有任务是否处理完。这里注意q.put(None)这个哨兵值，如果把None当作普通任务放入队列，消费者会把它也当成业务数据，所以需要在消费者内部显式判断并跳出。

3. 多进程核心：multiprocessing 的进程池与并发模型

3.1 为什么会需要多进程：绕开 GIL 的正确姿势

多线程在 I/O 密集型场景下表现良好，但一旦任务变成纯 CPU 计算，比如图像滤镜、矩阵乘法、哈希碰撞等，线程调度时争抢 GIL 的开销甚至会把性能拖到比单线程还低。multiprocessing模块正是为此存在的——它通过fork（Linux/macOS）或spawn（Windows）创建新进程，每个进程有独立的内存空间和独立的 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(target=heavy_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是传统 API，concurrent.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(processes=8) as pool: results = pool.map(is_prime, numbers) print(f"8 进程并行筛选耗时: {time.time() - start:.2f}s") print(f"质数个数: {sum(results)}")

Pool(processes=8)指定 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(target=append_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(target=producer, 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(target=busy_work, args=(100,)) for _ in range(10)] for t in threads: t.start() for t in threads: t.join() def run_process(): with Pool(processes=10) 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_workers=8) 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, savedir="downloads"): 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_ok=True) urls = [ f"https://example.com/files/{i}.zip" for i in range(20) ] with ThreadPoolExecutor(max_workers=8) 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 <pid>

Python 收到该信号后会在 stderr 输出各个线程的栈帧，通过最后等待的位置能判断是否卡在lock.acquire()上。faulthandler模块更规范地做这件事——在代码开头启用，运行时按Ctrl+C即打印线程栈：

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 = [f"https://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_workers=4) def blocking_db_query(user_id): import time time.sleep(0.5) return f"user_{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 并发编程真正需要掌握的技能。

本文还有配套的精品资源，点击获取