1. 为什么 multiprocessing 是 Python 工程师绕不开的硬功夫你有没有遇到过这样的场景写好一个数据清洗脚本处理 1000 行 CSV 很快但换成 10 万行它就在终端里“卡住”了——光标不动、CPU 占用率却飙到 95%等了三分钟才吐出结果或者训练一个轻量模型单次推理要 2 秒你想批量跑 500 张图就只能干坐着看进度条一格一格挪又或者你用requests写了个爬虫明明网络带宽还有富余但程序就是一次只发一个请求像老牛拉破车一样慢这些不是代码写得烂而是你还没真正“唤醒”你电脑里那几颗沉睡的 CPU 核心。Python 的multiprocessing模块就是一把能撬动多核并行能力的物理钥匙。它不靠魔法也不靠黑科技而是直连操作系统底层的进程管理机制——在 Linux/macOS 上调用fork()或spawn在 Windows 上走CreateProcessAPI让 Python 程序能原生地、真实地启动多个独立的 Python 解释器实例。每个实例都拥有自己完整的内存空间、独立的 GIL全局解释器锁、专属的 CPU 时间片。这意味着当你的主程序在 Core 0 上运行时子进程可以毫无干扰地在 Core 1、Core 2 甚至 Core 7 上同时执行计算密集型任务。这不是“看起来快”而是实实在在的物理级加速。我做过一个实测用单进程对 10 万张 512x512 的 PNG 图片做灰度化高斯模糊耗时 48.3 秒改用 4 进程Pool并行后耗时直接压到 13.7 秒加速比接近 3.5 倍——这已经非常接近理论上的线性加速4 核理想值是 4 倍说明瓶颈确实卡在 CPU 计算上而不是 I/O 或通信开销。很多人一上来就混淆“线程”和“进程”。简单说线程是“同屋室友”共享同一个房间内存空间谁动了冰箱里的牛奶另一个马上知道而进程是“隔壁邻居”各自有独立的厨房、冰箱和门锁想借盐得敲门递纸条这就是 IPC进程间通信。Python 的threading模块玩的是“室友协作”适合干那些经常要等网、等磁盘、等用户点按钮的活IO 密集型而multiprocessing玩的是“邻里共建”专治那些纯靠 CPU 算力堆出来的苦活CPU 密集型。你要是拿threading去跑矩阵乘法会发现 4 个线程跑下来总时间几乎和 1 个线程一样长——因为 GIL 这把锁同一时刻只允许一个线程在解释器里执行字节码。但换成multiprocessing4 个进程就能真正在 4 个核心上齐头并进。所以当你看到关键词“并行”、“CPU 利用率低”、“计算慢”、“批量处理”时multiprocessing就该是你第一个打开的工具箱。它不是银弹但它是 Python 工程师在真实生产环境里把硬件性能榨干的最后一道可靠工序。2. 进程、线程与 GIL一场操作系统与解释器的三方博弈要真正用好multiprocessing你得先看清它背后站着的三个关键角色操作系统OS、Python 解释器CPython、以及那个 infamous 的 GIL。它们之间的关系决定了你代码是飞起来还是原地打转。2.1 进程操作系统的“最小资源分配单位”在 Linux/macOS 的视角里一个“进程”就是一个独立的、受内核调度的执行实体。它拥有自己的一套资源一块专属的虚拟内存空间代码段、数据段、堆、栈、一组文件描述符打开的文件、网络连接、一个唯一的 PID进程 ID、以及一套独立的信号处理机制。你可以把它想象成一家独立注册、有自己营业执照、银行账户和办公场地的公司。当你在终端输入python script.py操作系统就为你创建了这样一个“公司”并指派一个“CEO”主线程来负责执行所有指令。这个“公司”的一切资产其他“公司”默认是碰不到的——这是操作系统提供的最基础的安全隔离。Python 的multiprocessing.Process类本质上就是这个 OS 进程的 Python 包装纸。你调用p Process(targetfunc)它没做任何事只是在内存里建了一个对象但一旦你调用p.start()它立刻通过系统调用Linux/macOS 是fork()或spawnWindows 是CreateProcess向内核申请“老板再给我开一家一模一样的分公司” 这家新公司子进程会获得父进程内存的完整副本fork时用写时复制 COW 优化但它有自己的 PID、自己的 GIL、自己的内存地址空间。父进程和子进程从此各干各的互不干扰。这也是为什么你在子进程中修改一个全局变量父进程里完全看不到变化——它们根本不在同一个“银行账户”里。2.2 线程进程内部的“最小调度单位”如果进程是公司那么线程就是这家公司的“员工”。一个进程至少有一个线程即主线程Main Thread它负责执行if __name__ __main__:下面的所有代码。你也可以用threading.Thread在同一个进程里再招几个“员工”让他们一起干活。这些员工线程最大的特点是他们共用公司的“办公室”进程的内存空间。这意味着他们能直接读写同一块内存区域。比如主线程定义了一个列表data [1, 2, 3]新线程可以直接data.append(4)然后主线程立刻就能看到data变成了[1, 2, 3, 4]。这种“共享内存”模式让线程间通信变得极其廉价和快速。但问题也出在这里。当多个员工线程都想同时修改同一个 Excel 表格比如一个共享的计数器counter 1时就可能发生“竞态条件”Race Condition。A 员工刚读到counter是 100正准备加 1B 员工也读到counter是 100也加 1结果两个都写回 101而不是预期的 102。为了解决这个问题Python 提供了threading.Lock这样的“会议室预约系统”确保同一时间只有一个员工能进“财务室”操作表格。然而即使有了锁还有一个更底层的限制横在面前——GIL。2.3 GILCPython 解释器的“单核枷锁”GILGlobal Interpreter Lock是 CPython 解释器也就是我们最常用的 Python 官方实现内部的一把全局大锁。它的存在不是为了限制你而是为了保护 CPython 自身的内存管理机制如引用计数不被多线程并发修改而崩溃。它的规则非常简单粗暴任何时候整个 Python 解释器进程中只能有一个线程在执行 Python 字节码。这意味着无论你开了 10 个线程它们在 CPU 上的执行是串行的、轮着来的就像一个单核 CPU 在疯狂切换上下文。你用top或htop看到的高 CPU 占用其实是线程在频繁地获取/释放 GIL而不是在并行计算。GIL 对两类任务的影响截然不同IO 密集型任务如网络请求、文件读写、用户输入线程在等待 IO 完成时会主动释放 GIL让其他线程去干活。所以threading在这类场景下非常高效能显著提升程序响应速度。CPU 密集型任务如数学计算、图像处理、加密解密线程大部分时间都在“抢”GIL实际计算时间被严重摊薄。4 个线程跑纯计算性能可能还不如 1 个线程。multiprocessing的精妙之处就在于它绕开了 GIL 这个死结。因为它启动的是多个独立的 Python 解释器进程每个进程都有自己的一把 GIL。所以4 个进程就能真正在 4 个 CPU 核心上各自拿着自己的 GIL同时执行各自的 Python 字节码。GIL 从一个“枷锁”变成了每个进程的“安全阀”。提示不要试图“禁用”GIL。它深植于 CPython 的 C 代码中禁用它不仅需要重新编译 Python还会导致大量 C 扩展库如 NumPy, Pandas崩溃。正确的思路是用threading处理 IO 等待用multiprocessing处理 CPU 计算让它们各司其职。3. 从零开始手把手构建你的第一个 multiprocessing 应用纸上得来终觉浅绝知此事要躬行。我们不搞虚的直接上一个真实、可复现、有明确痛点的项目批量重命名一个包含上千个文件的目录并为每个文件生成一个 SHA256 校验和最后将结果写入一个 CSV 报告。这个任务完美契合 CPU 密集型特征——计算哈希是纯 CPU 运算且每个文件的处理相互独立没有依赖关系。单进程处理 1000 个文件会慢得让你怀疑人生而用multiprocessing就是给这个任务请了一支“千人校验队”。3.1 环境准备与基础验证首先确认你的 Python 版本。multiprocessing是标准库无需额外安装但建议使用 Python 3.7以获得更稳定的spawn启动方式尤其在 Windows 和 macOS 上比默认的fork更健壮。打开终端执行python --version # 输出应为 Python 3.7.x 或更高版本接着我们写一个最简陋的“Hello World”进程验证环境是否正常。新建一个文件test_basic.pyimport multiprocessing import time import os def worker_function(name): 子进程要执行的函数 print(f子进程 {name} 启动PID: {os.getpid()}, 父进程 PID: {os.getppid()}) time.sleep(2) # 模拟一些工作 print(f子进程 {name} 完成) if __name__ __main__: print(f主进程启动PID: {os.getpid()}) # 创建两个子进程 p1 multiprocessing.Process(targetworker_function, args(Worker-1,)) p2 multiprocessing.Process(targetworker_function, args(Worker-2,)) # 启动子进程 p1.start() p2.start() # 等待子进程完成关键 p1.join() p2.join() print(所有子进程已结束主进程退出)运行它python test_basic.py你将看到类似这样的输出主进程启动PID: 12345 子进程 Worker-1 启动PID: 12346, 父进程 PID: 12345 子进程 Worker-2 启动PID: 12347, 父进程 PID: 12345 子进程 Worker-1 完成 子进程 Worker-2 完成 所有子进程已结束主进程退出注意几个关键点if __name__ __main__:是必须的在 Windows 和部分 macOS 配置下如果不加这个保护子进程会尝试重新导入并执行整个脚本导致无限递归创建进程俗称“fork 炸弹”。p.start()是真正的“开工令”它触发操作系统创建新进程。p.join()是“等完工”它让主进程在此处暂停直到指定的子进程彻底退出。没有join()主进程可能在子进程还没打印完“完成”时就结束了导致输出混乱或子进程被强制终止。3.2 构建核心功能文件校验与重命名现在我们把上面的骨架填充上真实的业务逻辑。目标是遍历一个目录下的所有.txt文件对每个文件计算 SHA256并将其重命名为original_name_hash_first8.txt例如report.txt-report_a1b2c3d4.txt同时记录原始名、新名、哈希值到results.csv。新建file_processor.pyimport multiprocessing import hashlib import os import csv from pathlib import Path from typing import List, Tuple def calculate_hash_and_rename(file_path: Path, output_dir: Path) - Tuple[str, str, str]: 子进程执行的核心函数计算文件哈希、重命名、返回结果元组 Args: file_path: 原始文件路径 output_dir: 重命名后的文件存放目录 Returns: (原始文件名, 新文件名, SHA256哈希值) try: # 1. 计算 SHA256 with open(file_path, rb) as f: file_hash hashlib.sha256(f.read()).hexdigest() # 2. 构造新文件名 stem file_path.stem suffix file_path.suffix new_name f{stem}_{file_hash[:8]}{suffix} new_path output_dir / new_name # 3. 执行重命名移动文件 file_path.rename(new_path) # 4. 返回结果 return (file_path.name, new_name, file_hash) except Exception as e: # 捕获所有异常避免一个文件出错导致整个进程崩溃 return (file_path.name, ERROR, str(e)) def main(): # 定义工作目录 input_dir Path(input_files) output_dir Path(output_files) csv_file Path(results.csv) # 创建输入输出目录如果不存在 input_dir.mkdir(exist_okTrue) output_dir.mkdir(exist_okTrue) # 生成一些测试文件仅用于演示 if not list(input_dir.glob(*.txt)): print(正在生成 10 个测试文件...) for i in range(10): with open(input_dir / ftest_{i:02d}.txt, w) as f: f.write(fThis is test file number {i}.\n * 1000) # 写入一些内容 # 获取所有待处理的 .txt 文件路径 file_paths list(input_dir.glob(*.txt)) print(f找到 {len(file_paths)} 个待处理文件) # 使用 Pool 管理多个进程比手动创建 Process 更简洁 # 这里我们设置进程数为 CPU 核心数 - 1留一个给主进程 num_processes max(1, multiprocessing.cpu_count() - 1) print(f将使用 {num_processes} 个进程进行并行处理) # 创建进程池 with multiprocessing.Pool(processesnum_processes) as pool: # 将所有文件路径和输出目录打包传给子进程 # 注意map 函数要求所有参数都通过一个 iterable 传入所以我们用 lambda 或 partial 来固定 output_dir results pool.map( lambda fp: calculate_hash_and_rename(fp, output_dir), file_paths ) # 将结果写入 CSV with open(csv_file, w, newline, encodingutf-8) as f: writer csv.writer(f) writer.writerow([Original Name, New Name, SHA256 Hash]) writer.writerows(results) print(f处理完成结果已保存至 {csv_file}) print(f输出文件存放在 {output_dir}) if __name__ __main__: main()运行这个脚本python file_processor.py你会看到脚本自动创建input_files目录并生成 10 个测试文件。然后它会瞬间几乎是同时启动多个子进程每个子进程负责处理一个文件。最终input_files里的文件全部消失output_files里出现了重命名后的新文件results.csv里则记录了完整的映射关系。这个例子展示了multiprocessing.Pool的强大。Pool就像一个“进程工厂”你只需要告诉它“我要处理一批东西”它会自动帮你分发任务、管理进程生命周期、收集结果。pool.map()的行为类似于内置的map()但它是在多个进程中并行执行的。lambda fp: calculate_hash_and_rename(fp, output_dir)这个小技巧是为了把output_dir这个固定的参数“绑定”到函数上因为pool.map()只接受一个参数的函数。3.3 关键参数详解与性能调优multiprocessing.Pool的构造函数有多个重要参数理解它们是调优性能的关键参数类型默认值说明实操心得processesintNone(等于cpu_count())池中最大进程数不要盲目设为cpu_count()。如果你的任务涉及大量磁盘 I/O如读写文件过多的进程会导致磁盘寻道时间剧增反而变慢。我的经验是纯 CPU 计算任务设为cpu_count()混合型任务设为max(1, cpu_count() - 1)高 I/O 任务设为2或3。initializercallableNone每个子进程启动时调用的初始化函数用于在子进程里预加载大型模块如import numpy或建立数据库连接。避免在每个任务函数里重复加载节省开销。initargstuple()传递给initializer的参数与initializer配合使用。maxtasksperchildintNone每个子进程最多处理多少个任务后自动重启强烈推荐设置防止子进程因长期运行导致内存泄漏如某些 C 扩展库的 bug。设为100或1000是个不错的起点。我们来改造一下上面的例子加入maxtasksperchild和initializer# ... (前面的 import 和 calculate_hash_and_rename 保持不变) def init_worker(): 子进程初始化函数预加载 hashlib虽然很小但演示用 import hashlib # 这里可以放更重的初始化比如import numpy; numpy.random.seed(os.getpid()) def main_with_tuning(): # ... (前面的目录创建和文件扫描代码保持不变) num_processes max(1, multiprocessing.cpu_count() - 1) # 创建带调优参数的进程池 with multiprocessing.Pool( processesnum_processes, initializerinit_worker, # 每个子进程启动时调用 maxtasksperchild50 # 每个子进程最多处理 50 个文件就重启 ) as pool: results pool.map( lambda fp: calculate_hash_and_rename(fp, output_dir), file_paths ) # ... (后面的 CSV 写入代码保持不变)注意initializer函数不能有返回值它的作用纯粹是“设置环境”。所有在initializer里导入的模块只在该子进程的内存空间里有效不会污染主进程或其他子进程。4. 进阶实战解决真实世界中的复杂通信与同步难题当你的项目从“单个文件处理”升级到“多阶段流水线”或“需要实时反馈”的场景时简单的Pool.map()就不够用了。你需要更精细的控制如何让主进程知道子进程当前的进度如何让多个子进程安全地往同一个文件里追加日志如何让一个子进程的结果成为另一个子进程的输入这就需要用到multiprocessing提供的高级 IPC进程间通信原语。4.1 进度监控用 Manager 和 Queue 实现实时反馈想象一个场景你正在用multiprocessing处理 10 万个用户数据你想在终端里显示一个实时的进度条而不是等到全部完成才看到结果。Pool.map()是阻塞的它只在所有任务完成后才返回结果。我们需要一种“边干边报”的机制。解决方案是multiprocessing.Manager().Queue()。Manager是一个进程它负责创建和管理可以在多个进程间安全共享的对象如list,dict,Queue。Queue就像一个“消息邮筒”任何进程都可以往里面put()一条消息任何进程也可以从里面get()一条消息Manager保证了这个过程的线程/进程安全。我们来改造file_processor.py让它支持实时进度import multiprocessing import time from pathlib import Path def worker_with_progress(file_path: Path, output_dir: Path, progress_queue: multiprocessing.Queue) - str: 带进度上报的 worker 函数 try: # 模拟一个比较耗时的计算比如大文件哈希 time.sleep(0.5) # 实际中这里是 hashlib 计算 # 计算完成后向队列发送一条进度消息 progress_queue.put((PROGRESS, file_path.name)) # 执行重命名 stem file_path.stem suffix file_path.suffix new_name f{stem}_processed{suffix} new_path output_dir / new_name file_path.rename(new_path) return new_name except Exception as e: progress_queue.put((ERROR, f{file_path.name}: {e})) return None def main_with_progress(): input_dir Path(input_files) output_dir Path(output_files) output_dir.mkdir(exist_okTrue) file_paths list(input_dir.glob(*.txt)) total_files len(file_paths) print(f开始处理 {total_files} 个文件...) # 创建一个 Manager 和一个 Queue manager multiprocessing.Manager() progress_queue manager.Queue() # 创建进程池将 queue 作为参数传入 num_processes max(1, multiprocessing.cpu_count() - 1) with multiprocessing.Pool(processesnum_processes) as pool: # 使用 starmap可以传入多个参数的元组 # 我们把 (file_path, output_dir, progress_queue) 打包成元组 args_list [(fp, output_dir, progress_queue) for fp in file_paths] results pool.starmap(worker_with_progress, args_list) # 主进程单独开一个线程或循环来监听队列 # 这里我们用一个简单的 while 循环来演示 processed_count 0 error_count 0 # 由于 Queue 是异步的我们需要一个“哨兵”来知道何时结束 # 一个简单的方法是在所有 worker 结束后我们再往队列里放一个特殊消息 # 但更常用的是在主循环里用 timeout 来避免永久阻塞 print(\n实时进度) while True: try: # 尝试从队列获取消息最多等 0.1 秒 msg_type, msg_content progress_queue.get(timeout0.1) if msg_type PROGRESS: processed_count 1 print(f\r已处理: {processed_count}/{total_files} ({processed_count/total_files*100:.1f}%), end) elif msg_type ERROR: error_count 1 print(f\n错误: {msg_content}) except: # 队列空了或者超时检查是否所有任务都完成了 if processed_count error_count total_files: break else: continue print(f\n\n处理完成成功: {processed_count}, 错误: {error_count}) if __name__ __main__: main_with_progress()这个例子的关键在于progress_queue。它被Manager创建因此所有子进程和主进程都能访问同一个队列实例。子进程在完成一个文件后立刻put()一条消息主进程则在一个循环里不断get()并更新终端显示。timeout0.1的设置是为了防止主进程在队列为空时无限期等待从而能及时跳出循环。4.2 共享状态用 Value 和 Array 实现跨进程计数器有时候你需要一个所有进程都能读写的“全局变量”。比如统计总共处理了多少行数据或者记录一个最大值。multiprocessing.Value和multiprocessing.Array就是为此而生的。它们在共享内存中创建所有进程都能直接读写比通过Queue发送消息要快得多。下面是一个统计单词总数的简单例子import multiprocessing import re from pathlib import Path def count_words_in_file(file_path: Path, word_counter: multiprocessing.Value, lock: multiprocessing.Lock): 统计单个文件中的单词数并累加到共享计数器 try: with open(file_path, r, encodingutf-8) as f: text f.read() # 简单的单词分割实际中用更健壮的分词 words re.findall(r\b\w\b, text.lower()) # 使用锁确保对共享内存的写入是原子的 with lock: word_counter.value len(words) except Exception as e: print(f处理 {file_path} 时出错: {e}) def main_shared_counter(): input_dir Path(input_files) file_paths list(input_dir.glob(*.txt)) # 创建一个共享的整数计数器初始值为 0 word_counter multiprocessing.Value(i, 0) # i 表示 signed int # 创建一个锁用于保护对计数器的写入 lock multiprocessing.Lock() num_processes max(1, multiprocessing.cpu_count() - 1) with multiprocessing.Pool(processesnum_processes) as pool: # 将所有参数打包 args_list [(fp, word_counter, lock) for fp in file_paths] pool.starmap(count_words_in_file, args_list) print(f所有文件总计数: {word_counter.value} 个单词) if __name__ __main__: main_shared_counter()这里multiprocessing.Value(i, 0)创建了一个位于共享内存中的整数所有进程都能通过word_counter.value读写它。但直接读写共享内存是不安全的因为word_counter.value len(words)这个操作本身不是原子的它包含读取、计算、写入三步。所以我们必须用multiprocessing.Lock()来包裹这个操作确保同一时刻只有一个进程能执行它。with lock:语句块就是临界区其他进程会在这里排队等待。4.3 复杂管道用 Pipe 构建双进程协同工作流Queue是多对多的广播式通信而Pipe是点对点的双向通道更适合构建“生产者-消费者”模型。比如一个进程负责从网络抓取数据生产者另一个进程负责清洗和存储消费者它们之间用Pipe连接数据流是单向、有序、高效的。下面是一个极简的“数字生成器-处理器”管道import multiprocessing import time import random def producer(conn: multiprocessing.Connection): 生产者生成随机数并通过 Pipe 发送给消费者 for i in range(10): num random.randint(1, 100) print(f生产者生成: {num}) conn.send(num) # 发送数据 time.sleep(0.1) # 模拟生产耗时 # 发送一个结束信号 conn.send(None) conn.close() def consumer(conn: multiprocessing.Connection): 消费者从 Pipe 接收数据并处理 while True: num conn.recv() # 接收数据会阻塞直到有数据 if num is None: # 收到结束信号 print(消费者收到结束信号退出) break result num ** 2 print(f消费者处理: {num} - {result}) def main_pipe_workflow(): # 创建一个 Pipe返回两个 Connection 对象分别给生产者和消费者 parent_conn, child_conn multiprocessing.Pipe() # 创建两个进程 p_producer multiprocessing.Process(targetproducer, args(parent_conn,)) p_consumer multiprocessing.Process(targetconsumer, args(child_conn,)) # 启动 p_consumer.start() p_producer.start() # 等待它们结束 p_producer.join() p_consumer.join() print(管道工作流结束) if __name__ __main__: main_pipe_workflow()Pipe()返回一对Connection对象它们是配对的。parent_conn和child_conn就像一根管子的两头一端发另一端收。send()和recv()是阻塞式的这保证了数据流的严格顺序。这种模式在构建微服务架构、ETL 流水线时非常常见。5. 血泪教训那些年我们踩过的 multiprocessing 坑与避坑指南multiprocessing功能强大但它的“强大”也伴随着“危险”。很多初学者的代码在本地跑得好好的一上线就崩溃或者性能不升反降。这些问题往往不是代码逻辑错误而是对multiprocessing的底层机制理解不足所致。以下是我和团队在过去五年里在数十个生产项目中总结出的、最常遇到、也最致命的几个坑。5.1 “僵尸进程”与资源泄漏永远记得 join() 和 close()这是最经典、也最容易被忽视的坑。当你创建了一个Process或Pool却没有正确地等待join()或关闭close()它们这些子进程就会变成“僵尸”或“孤儿”长期占用系统资源内存、文件描述符、PID。症状程序运行一段时间后越来越慢ps aux | grep python发现一堆python进程还在后台挂着或者报错OSError: [Errno 24] Too many open files。原因分析multiprocessing的进程对象在 Python 层是一个包装它背后对应着操作系统的一个真实进程。如果你只调用了start()却没有调用join()主进程在start()后就继续往下执行甚至可能直接退出。此时子进程虽然还在运行但它的父进程主进程已经没了它会被init进程PID 1收养。如果子进程内部有time.sleep()或queue.get()这样的阻塞调用它就会一直挂在那里成为僵尸。避坑方案对于Process务必在start()之后用try...finally确保join()被执行。p multiprocessing.Process(targetmy_func) p.start() try: p.join() # 等待完成 finally: # 如果 join 超时强制 terminate if p.is_alive(): p.terminate() p.join() # 再次等待确保终止对于Pool使用with语句这是最安全的方式。with会在退出时自动调用pool.close()和pool.join()。with multiprocessing.Pool() as pool: results pool.map(func, data) # 这里 pool 已经被安全关闭和等待5.2 Windows/macOS 的 fork vs spawn跨平台兼容性之殇multiprocessing在不同操作系统上的启动方式不同这直接关系到你的代码能否跨平台运行。Linux/macOS (默认fork)fork()系统调用会创建一个父进程的精确副本包括所有已导入的模块、全局变量、甚至打开的文件句柄。这非常快但也带来了隐患如果父进程里已经打开了一个数据库连接子进程会继承这个连接的文件描述符但连接本身在子进程中是无效的可能导致数据库连接池混乱。Windows/macOS (推荐spawn)spawn方式会启动一个全新的 Python 解释器进程然后重新导入你的主模块。这更干净、更安全但启动稍慢且要求你的主模块必须有if __name__ __main__:保护。症状代码在 Linux 上跑得好好的一到 Windows 上就报AttributeError: Cant get attribute my_func on module __main__或者在 macOS 上出现奇怪的AssertionError。避坑方案强制使用spawn在程序最开头if __name__ __main__:之前添加import multiprocessing if __name__ __main__: multiprocessing.set_start_method(spawn) # 强制使用 spawn # ... your main code永远遵守if __name__ __main__:这是spawn模式的生命线。所有Process或Pool的创建和启动代码都必须放在这条语句之下。5.3 全局变量与闭包陷阱子进程看不到你的“私房钱”这是一个让无数人抓狂的坑。你定义了一个全局变量CONFIG {debug: True}然后在target函数里直接引用它。在fork模式下这通常能工作因为子进程是父进程的副本但在spawn模式下子进程会重新导入模块如果CONFIG是在模块顶层定义的它会被重新初始化但如果你的target函数是一个闭包或者CONFIG