Python 多线程与多进程并发深度解析

1. 背景：为什么 Python 并发总是"又爱又恨"

Python 的并发话题几乎是每个工程师绕不开的争论场：有人说"Python 多线程是假的，被 GIL 锁死了"，也有人说"Python 里跑并发就是自找麻烦，老老实实串行吧"。这两种说法都只对了一半。

真实情况是：Python 的并发工具箱里既有"线程"（threading），也有"进程"（multiprocessing），还有一层优雅的封装（concurrent.futures），它们各有各的适用域。问题从来不是"Python 能不能并发"，而是"在什么场景下选哪个，以及怎么避开那些藏在标准库深处的坑"。

对工业数采场景而言，并发的需求几乎是与生俱来的：

• 多设备并行采集：一条产线上可能有几十台 CNC、PLC、扫码枪、传感器网关，串行轮询一个周期动辄几秒，一旦超过采集节拍就会丢数据；
• I/O 密集的上报与存储：采集到的点位数据要写数据库、发 Kafka、走 MQTT，这些操作大部分时间都花在等待网络/磁盘 I/O 上，串行执行等于把 CPU 空转的时间全部浪费掉；
• 批量离线任务：日报分析、质量报表、模型批处理，往往要处理几十万行数据，单进程跑半天，多进程一拆就是分钟级；
• 边缘 AI 推理：推理前后要做预处理/后处理（NumPy 操作），推理本身又可能调用 C 扩展或外部服务，需要把"CPU 密集"与"I/O 等待"两类任务分流。

为什么选择"线程/进程"而不是直接上 asyncio？因为工业数采的现实是：大量存量 SDK 和第三方库是同步阻塞的（比如通过 ctypes 调 C 库、调厂商 DLL、走 requests 同步 HTTP），asyncio 无法让它们"飞"起来；而线程/进程模型与同步代码天然兼容，改造成本最低。asyncio 适合"自己掌控一切 I/O"的新代码，threading/multiprocessing 适合"存量同步代码并行化"，两者不是替代关系而是互补关系。

一句话定位本篇：把 CPython 的线程与进程机制讲到骨子里，让你在工业数采与通用 Python 工程中，知道"什么时候该用哪个、代码怎么写才对、踩坑了怎么定位"。

2. 核心概念速览：线程、进程、GIL 与并发/并行

2.1 并发（Concurrency）与并行（Parallelism）

这是两个被滥用最多的词：

关键认知：并发是"结构"，并行是"物理"。Python 多线程解决的是"并发结构"问题（把阻塞等待藏起来），多进程解决的是"并行算力"问题（把计算摊到多核上）。别指望用线程去加速纯 CPU 计算——那是进程的活。

2.2 线程（Thread）与进程（Process）

2.3 GIL（Global Interpreter Lock）：绕不开的"解释器大锁"

GIL 是 CPython（官方解释器）的全局解释器锁：同一时刻，一个进程内只有一个线程能执行 Python 字节码。这不是 bug，是历史设计——为了保证 CPython 的引用计数内存管理在单线程模型下简单高效，代价就是牺牲多线程的 CPU 并行能力。

GIL 的三个真相：

1. I/O 密集不受 GIL 影响：线程在等待 I/O（recv、read、sleep、requests 等）时，会主动释放 GIL，其他线程可以拿到锁继续跑。所以 100 个线程做网络请求，吞吐照样能上去。
2. CPU 密集被 GIL 卡死：纯 Python 计算（for 循环、字符串处理、纯 Python 算法）每个线程拿到 GIL 后只能跑一小段时间（约 5ms 的切换间隔，sys.getswitchinterval() 可查），多线程跑 CPU 密集任务不仅不加速，还可能因锁竞争反而变慢。
3. C 扩展可以绕开 GIL：NumPy、PyTorch、re、zlib 等底层 C 实现的函数会主动释放 GIL（它们操作的是自己的数据，不碰 Python 对象），所以 numpy 数组运算、torch 推理在多线程下依然可以并行加速。

换句话说：线程负责"等得起"，进程负责"算得起"。判断依据就一句话：任务瓶颈在"等"（I/O）还是在"算"（CPU）。

2.4 同步原语分类表

3. API 说明：三大并发模块全家桶

3.1 threading：OS 线程标准库

核心类与函数：

锁的上下文管理器（强烈推荐）：

lock = threading.Lock() with lock: # 等价 acquire + try/finally release # 临界区

3.2 queue：线程安全队列

queue.Queue 内部用 threading.Condition + deque 实现，是线程间传递数据的首选，比手写 list + lock 安全得多。

3.3 multiprocessing：多进程标准库

核心类与函数：

Windows 的关键差异：Windows 没有 fork，子进程只能 spawn——即重新启动一个 Python 解释器，导入主模块，再执行目标函数。因此创建 Process 的代码必须放在 if __name__ == "__main__": 里，否则会无限递归创建进程。这也是工业数采项目在 Windows 上跑多进程最容易踩的第一坑。

3.4 concurrent.futures：统一线程池/进程池封装（推荐入口）

为什么优先用 concurrent.futures？因为它把"线程 vs 进程"的选择收敛成一个单词的差异：把 ThreadPoolExecutor 换成 ProcessPoolExecutor，其余代码几乎不变。这让你可以先写出正确的并发逻辑，再根据实测性能决定用线程还是进程，把"选型"从编码期推迟到调优期。

内置 API 的一个黄金搭档：concurrent.futures 内部就是 queue + 工作线程/进程实现的，submit 的任务排队执行，max_workers 控制并发度，天然限流，不用手写"线程池"。

4. 详细使用：9 组可运行示例

以下示例均为完整可运行代码（Python 3.8+），建议在 if __name__ == "__main__": 保护下运行，Windows 与 Linux 通用。

4.1 最小线程闭环：多线程并发 I/O

import threading import time import requests def fetch(url: str, results: dict): start = time.time() resp = requests.get(url, timeout=5) # I/O 密集：等待期间释放 GIL results[url] = (resp.status_code, round(time.time() - start, 2)) if __name__ == "__main__": urls = [f"https://httpbin.org/delay/1" for _ in range(8)] results = {} threads = [threading.Thread(target=fetch, args=(u, results)) for u in urls] for t in threads: t.start() for t in threads: t.join() print(results) # 8 个请求约 1.x 秒完成，串行则约 8 秒

要点：requests.get 是同步阻塞调用，8 个线程并行等待网络，总耗时≈单个请求耗时，吞吐提升约 8 倍。这是"线程解决 I/O 并发"的最佳写照。

4.2 锁的正确姿势：保护共享计数器

import threading class SafeCounter: def __init__(self): self._count = 0 self._lock = threading.Lock() def incr(self, n: int = 1): with self._lock: # 必须用 with，异常也能释放 self._count += n def value(self) -> int: with self._lock: # 读也要加锁！ return self._count if __name__ == "__main__": c = SafeCounter() def worker(): for _ in range(100_000): c.incr() ts = [threading.Thread(target=worker) for _ in range(8)] [t.start() for t in ts] [t.join() for t in ts] print(c.value()) # 800_000，不加锁会明显小于该值

要点：+= 不是原子操作（LOAD → ADD → STORE 三步），多线程下会丢更新。读和写都要加锁，否则可能读到中间状态。这是"共享状态并发"的黄金法则。

4.3 条件变量：生产者-消费者

import threading import queue import time import random q = queue.Queue(maxsize=20) # 有界队列：天然背压 stop_event = threading.Event() def producer(name: str): for i in range(30): if stop_event.is_set(): break item = f"{name}-{i}" q.put(item) # 满时阻塞，天然节流 time.sleep(random.uniform(0.02, 0.08)) q.put(None) # 哨兵：结束信号 def consumer(): while True: item = q.get() if item is None: # 收到哨兵退出 q.task_done() break time.sleep(random.uniform(0.01, 0.05)) # 模拟处理 print(f"consume {item}") q.task_done() if __name__ == "__main__": c = threading.Thread(target=consumer, daemon=True) c.start() producers = [threading.Thread(target=producer, args=(f"P{i}",)) for i in range(3)] [p.start() for p in producers] [p.join() for p in producers] q.join() # 等待所有任务处理完 print("done")

要点：queue.Queue 自带阻塞与线程安全，生产者消费者用队列 + 哨兵（None）是标准范式。有界队列（maxsize）让"生产过快"自然变成阻塞，避免内存被撑爆。

4.4 concurrent.futures 线程池：批量 HTTP 上报

from concurrent.futures import ThreadPoolExecutor, as_completed import requests import time def upload(payload: dict) -> tuple: # 模拟向采集平台上报点位数据 resp = requests.post("https://httpbin.org/post", json=payload, timeout=5) return payload["device_id"], resp.status_code if __name__ == "__main__": payloads = [{"device_id": f"CNC-{i:03d}", "ts": time.time(), "value": i * 0.5} for i in range(100)] ok = 0 with ThreadPoolExecutor(max_workers=16) as ex: # 16 路并发上报 futures = {ex.submit(upload, p): p["device_id"] for p in payloads} for fut in as_completed(futures): device_id, status = fut.result() # 异常在此抛出 if status == 200: ok += 1 print(f"success: {ok}/{len(payloads)}")

要点：with 块退出时自动 shutdown(wait=True) 等待全部任务完成。as_completed 让"先完成的先处理"，适合结果不依赖顺序的场景。future.result() 会把任务里的异常原样抛出，必须在循环里 try/except。

4.5 concurrent.futures 进程池：CPU 密集加速

from concurrent.futures import ProcessPoolExecutor import time def heavy_calc(n: int) -> int: # 纯 Python CPU 密集：计算第 n 个素数 count, num = 0, 2 while count < n: for d in range(2, int(num ** 0.5) + 1): if num % d == 0: break else: count += 1 num += 1 return num - 1 if __name__ == "__main__": targets = list(range(1000, 1100)) # 100 个素数计算任务 start = time.time() with ProcessPoolExecutor(max_workers=8) as ex: # 8 进程真并行 results = list(ex.map(heavy_calc, targets, chunksize=10)) elapsed = time.time() - start print(f"8 processes: {elapsed:.2f}s, first={results[0]}, last={results[-1]}") # 对比线程池（会被 GIL 卡住，往往更慢）： from concurrent.futures import ThreadPoolExecutor start = time.time() with ThreadPoolExecutor(max_workers=8) as ex: results_t = list(ex.map(heavy_calc, targets, chunksize=10)) print(f"8 threads: {time.time() - start:.2f}s")

要点：纯 Python CPU 密集任务，进程池显著快于线程池（实测常为 4~7 倍差距，受 GIL 切换影响线程池甚至可能比串行慢）。chunksize 让每个进程一次领取一批任务，减少 IPC 次数，任务量大时必须设置。

4.6 multiprocessing：进程间队列与共享值

import multiprocessing as mp def worker(q: mp.Queue, counter: mp.Value): for i in range(100): q.put(i) with counter.get_lock(): # 共享值必须加锁 counter.value += 1 q.put(None) # 哨兵 if __name__ == "__main__": ctx = mp.get_context("spawn") # Windows 显式声明 spawn q = ctx.Queue() counter = ctx.Value("i", 0) # 有符号 int，初始 0 procs = [ctx.Process(target=worker, args=(q, counter)) for _ in range(4)] [p.start() for p in procs] for _ in range(4): # 每个进程一个哨兵 while True: item = q.get() if item is None: break [p.join() for p in procs] print("total items:", counter.value) # 400

要点：进程间通信必须用 Queue/Pipe/Value/Array，不能直接共享普通 Python 对象（每个进程有独立内存）。Value 的读写要 get_lock()，否则同样有竞态。get_context("spawn") 显式统一启动方式，避免 Windows/Linux 行为差异。

4.7 线程局部存储：每个线程独立的日志上下文

import threading import time local = threading.local() def worker(device_id: str): local.device_id = device_id # 每个线程独立的"私有变量" for i in range(3): print(f"[{threading.current_thread().name}] " f"device={local.device_id} sample={i}") time.sleep(0.05) if __name__ == "__main__": ts = [threading.Thread(target=worker, args=(f"CNC-{i}",), name=f"T{i}") for i in range(4)] [t.start() for t in ts] [t.join() for t in ts]

要点：threading.local() 是"线程私有命名空间"，各线程读写互不干扰，适合存放会话、连接、日志上下文等"每线程一份"的对象，从根上避免共享冲突。

4.8 半开连接检测：守护线程 + Event 优雅关闭

import threading import time def monitor(stop_event: threading.Event): """守护线程：每秒检查一次全局状态""" while not stop_event.is_set(): print("monitor tick", time.strftime("%H:%M:%S")) stop_event.wait(timeout=1.0) # 可中断的 sleep def main_loop(): stop = threading.Event() m = threading.Thread(target=monitor, args=(stop,), daemon=True) m.start() try: for i in range(5): print("main work", i) time.sleep(0.5) finally: stop.set() # 通知守护线程退出 m.join(timeout=2.0) print("clean exit") if __name__ == "__main__": main_loop()

要点：守护线程 + Event 是 Python 服务优雅关闭的标准组合。event.wait(timeout) 让线程"睡"得可被唤醒，避免 time.sleep() 无法及时响应停止信号。主流程用 try/finally 保证无论如何都发出停止信号。

4.9 工业数采：多设备并行采集 + 落库消费者

"""模拟工业数采网关：N 台设备并行采集，统一进入队列，单消费者批量落库。""" import threading import queue import time import random BATCH_SIZE = 50 q = queue.Queue(maxsize=1000) def collect_device(device_id: str): """设备采集线程：周期读取点位（模拟阻塞 I/O）""" seq = 0 while True: seq += 1 # 模拟 PLC/CNC 读取（真实场景为 snap7/FOCAS/MC 等同步 SDK 调用） time.sleep(random.uniform(0.05, 0.15)) q.put({"device": device_id, "seq": seq, "ts": time.time(), "value": random.uniform(0, 100)}) def batch_store(): """落库消费者：攒够一批再批量写库，降低事务开销""" buffer = [] while True: item = q.get() buffer.append(item) if len(buffer) >= BATCH_SIZE: # 批量写 TDengine / Kafka / MySQL（示例只打印） print(f"flush {len(buffer)} rows, " f"first={buffer[0]['device']}:{buffer[0]['seq']} " f"last={buffer[-1]['device']}:{buffer[-1]['seq']}") buffer.clear() q.task_done() if __name__ == "__main__": devices = [f"CNC-{i:02d}" for i in range(10)] threads = [threading.Thread(target=collect_device, args=(d,), daemon=True) for d in devices] consumer = threading.Thread(target=batch_store, daemon=True) consumer.start() [t.start() for t in threads] try: time.sleep(10) # 网关运行 10 秒 except KeyboardInterrupt: pass print(f"queue size at exit: {q.qsize()}")

要点：设备线程群只负责"读"，落库线程只负责"写"，中间用有界队列解耦。消费者攒批写入，把 N 次小事务合并成 1 次批量事务。守护线程保证主程序退出时不用等它们。这就是工业数采网关"采集层 / 缓冲层 / 存储层"三层的并发骨架。

5. 底层实现剖析：GIL、线程切换与进程模型

5.1 GIL 的调度与切换

CPython 的 GIL 采用固定时间片 + 信号切换策略：

• 当前线程持有 GIL 执行字节码，每执行一定数量的指令（sys.getswitchinterval()，默认约 5ms）或遇到"检查点"（如函数调用、循环回跳），就检查是否有其他线程在等待 GIL；
• 若有人等待，当前线程释放 GIL 并切换；
• 释放期间，等待线程通过信号量机制竞争获取 GIL，切换成本约几十微秒到几百微秒（含上下文切换与缓存失效）。

关键点：GIL 切换是"协作式 + 抢占式"混合。纯 Python 代码会被时间片打断（所以多线程纯计算不会真正并行）；但进入 C 扩展时，若该扩展声明"我在做不涉及 Python 对象的计算，请释放 GIL"（如 Py_BEGIN_ALLOW_THREADS），其他线程就能插进来跑——这正是 NumPy 多线程依然能加速的原因。

5.2 为什么 I/O 密集线程并发有效

requests.get、socket.recv、time.sleep、文件读写等阻塞调用，在 CPython 内部都会走到"释放 GIL → 阻塞等待 → 事件就绪 → 重新抢 GIL"的路径。因此：

线程 A：拿到 GIL → 发 HTTP 请求 → 释放 GIL → 等网络（0.5s） 线程 B：抢到 GIL → 发 HTTP 请求 → 释放 GIL → 等网络（0.5s） ...

N 个线程的等待期完全重叠，总耗时≈单个请求耗时。这就是"线程解决 I/O 并发"的底层原理。

5.3 多进程为什么能绕开 GIL

每个 multiprocessing 子进程都是独立的 CPython 解释器，各自持有一把 GIL，互不干扰。N 个进程 = N 把 GIL = N 个真并行执行流（只要核够多）。

代价是：

1. 启动开销大：spawn 方式要重新导入解释器和主模块（Windows 上可能几百毫秒）；
2. 通信开销大：进程间数据要 pickle 序列化 + 管道传输，大数据结构频繁传递会拖垮性能；
3. 内存翻倍：每个进程一份解释器与全局对象（默认 copy-on-write 在 spawn 下不存在）。

所以进程池适合"任务大、数据小、通信少"的粗粒度并行；线程池适合"任务碎、共享多、I/O 为主"的细粒度并发。

5.4 三种进程启动方式

工业数采建议：统一用 spawn（或 get_context("spawn")），保证 Windows 开发环境与 Linux 服务器行为一致，避免"本地能跑、服务器死锁"的经典翻车。

5.5 concurrent.futures 内部结构

• ThreadPoolExecutor：内部是一个工作线程集合 + 一个 queue.Queue。submit 把任务（闭包 + 参数）放进队列，空闲工作线程取出执行，结果写入 Future；
• ProcessPoolExecutor：内部是一个进程集合 + 两个队列（任务队列 + 结果队列）+ 每进程一个"call queue"线程，任务经 pickle 发给子进程，结果 pickle 传回；
• Future 对象：持有"任务状态"（pending/running/done）、结果或异常、一组回调，result() 阻塞等待状态变为 done。

理解这个结构，你就明白为什么"任务参数必须可 pickle"（进程池）、"回调里不能做重活"（线程池工作线程有限）、"池对象要 with 管理"（退出时排空队列）。

6. 常错点 26 坑

7. 性能优化清单 12 条

1. 先分类再选型：任务瓶颈是"等"（网络/磁盘/串口）→ 线程；是"算"（纯 Python 循环）→ 进程或 C 扩展。
2. 永远用线程池/进程池：ThreadPoolExecutor/ProcessPoolExecutor 自带限流与复用，别手写线程数组。
3. 把 C 扩展留给线程：NumPy/PyTorch/re 等会释放 GIL，多线程调用它们可并行加速且无 pickle 开销。
4. 进程池任务粗粒度 + chunksize：任务数远大于进程数时设置 chunksize 批量领取，减少 IPC 次数。
5. 有界队列 + 背压：maxsize 设定上限，生产者满则阻塞，防止内存爆炸（工业数采队列满时应"丢旧保新 + 告警"而非无限堆积）。
6. 批量替代细粒度：攒批写库/上报，把 N 次小事务合并为 1 次批量事务，吞吐提升数倍。
7. threading.local 隔离连接：每线程独立 DB/HTTP 连接，避免共享连接锁竞争。
8. 锁粒度最小化：能用原子操作（queue、Value 加锁）不用大锁；锁内绝不做 I/O。
9. 避免回调重活：add_done_callback 只做信号/入队，重活重新 submit。
10. 进程初始化器：ProcessPoolExecutor(initializer=init, initargs=...) 每进程一次创建连接/加载模型，避免每任务重复开销。
11. 测量驱动：cProfile 看 CPU、time 对比串行基线，别拍脑袋选型。
12. 监控线程健康：定期检查队列长度、存活线程数、Future 异常率，接 Prometheus 计数。

8. 工业数采落地建议

8.1 采集网关并发模型选型表

8.2 数据流架构模板

设备线程群（每设备 1 线程） │ 同步读点（snap7/FOCAS/MC/Modbus TCP） ▼ 有界队列 queue.Queue(maxsize=1000) ← 缓冲层：解耦、背压 │ ▼ 消费者线程（1~2 个）→ 攒批 → 批量写 TDengine/Kafka/MQTT │ ▼ 监控：队列长度 / 线程存活 / 批处理耗时 → Prometheus 指标

8.3 工程红线五条

1. 共享数据三选一：要么只读初始化后不再写，要么加锁，要么走队列——不允许"裸改"。
2. 异常统一出口：采集线程 catch 异常后记录日志 + 计数告警，绝不让一个设备异常静默吞掉整条链路。
3. 队列满必须可见：put_nowait + 满时丢旧保新 + 丢包计数，宁可丢旧数据也要保证实时性可观测。
4. Windows 开发、Linux 部署：全程 spawn 上下文，避免 fork 死锁与平台差异。
5. 先正确后优化：先线程池写出正确链路，再 profile 决定是否上进程池/无锁结构。

9. 总结五要点

1. 判断句：任务瓶颈在"等"用线程，在"算"用进程；GIL 只锁 Python 字节码，不锁 I/O 等待，也不锁释放了 GIL 的 C 扩展。
2. 正确姿势：并发入口永远优先 concurrent.futures 的线程池/进程池——一个单词切换模型，自带限流与复用。
3. 通信铁律：线程间用 queue.Queue，进程间用 multiprocessing.Queue/Pipe/Value；共享状态必须加锁，读也要加锁。
4. 平台一致性：多进程全程用 spawn 上下文，创建代码放 if __name__ == "__main__": 保护块。
5. 工程落地：工业数采用"设备线程群 → 有界队列 → 批量落库消费者"三层骨架，队列满丢旧保新 + 计数告警，先正确后优化。

10. FAQ 12 条速查

Q1：Python 多线程到底是不是假的？不是。多线程对 I/O 密集任务是真实有效的（等待期释放 GIL）；只有纯 CPU 计算受 GIL 限制无法并行。把"多线程"与"CPU 并行"混为一谈才是误区。

Q2：GIL 什么时候会被移除？CPython 官方有 free-threading（no-GIL）实验分支（3.13t 构建），但生产环境主流仍是 GIL 版本。设计并发代码时不必赌未来，线程/进程双模型足够应对现状。

Q3：线程池和进程池怎么选？参数能否 pickle？数据量大不大？任务是 I/O 还是 CPU？——进程池要求任务与参数可 pickle 且数据量小，否则选线程池。多数 I/O 场景线程池就够。

Q4：q.join() 和 q.task_done() 是什么关系？put 一次计数 +1，task_done 一次计数 -1，join 阻塞直到计数归零。每 get 一次必须对应 task_done 一次，否则 join 卡死。

Q5：为什么我的进程池比串行还慢？多半是任务太小/数据太大，pickle + IPC 开销盖过了并行收益。加大任务粒度、减小传参体积、设 chunksize。

Q6：Windows 上跑 multiprocessing 报 "An attempt has been made to start a new process..."？创建 Process 的代码没放在 if __name__ == "__main__": 里，spawn 重新导入主模块时递归创建了子进程。放进主保护块即可。

Q7：多线程读同一个文件安全吗？读同一只读文件安全；写同一文件不安全（偏移量竞争）。多线程写文件用 logging 或串行队列。

Q8：requests.Session 线程安全吗？官方不建议跨线程共享 Session。每线程建 Session，或用 requests 的连接池模式自行管理。

Q9：Event.wait(timeout) 与 time.sleep(timeout) 差别？Event.wait 可被 set() 立即唤醒，time.sleep 只能等完。守护线程退出信号用 Event。

Q10：怎么判断我的代码真在并行？multiprocessing 用 psutil 看 CPU 占用；threading 用 sys.setswitchinterval 观察切换，或对纯 CPU 任务对比串行耗时（线程不会更快）。

Q11：进程池能传自定义类实例吗？能，但类必须模块级可导入且可 pickle。lambda、嵌套函数、锁对象不能传。用 Manager 可共享 dict/list 等代理对象。

Q12：工业数采网关线程数设多少？I/O 密集可放宽（设备数 + 少量余量，如设备 20 台 → 24 线程）；CPU 密集用 os.cpu_count() 或略减；混合负载两类任务拆两个池。