Python 分布式进程：把任务派到"分店"去执行
引言：一家餐厅忙不过来，那就开分店
前面我们讲了多进程（一家店里雇多个厨师）和多线程（一个厨房里多个厨师配合）。但无论怎么优化，一台机器的性能总有上限——CPU 核心数固定、内存容量固定、网络带宽固定。
如果生意好到一家店完全接不过来怎么办？
答案很简单：开分店。
- 总店（Master）负责接单、派单；
- 分店（Worker）负责做菜、上菜；
- 总店和分店之间通过电话/网络沟通。
这就是分布式进程——把任务从一台机器分发到多台机器上执行，突破单机性能瓶颈。
Python 的 multiprocessing.managers 模块把这一切封装得极其简单：你不需要懂网络编程，不需要写 socket 代码，只需要像操作本地队列一样操作"网络队列"，就能实现分布式计算。
这一篇，我们从单机多进程的局限讲起，逐步引出分布式进程的概念，再手把手教你用 managers 模块搭建一个跨机器的 Master-Worker 系统。
一、为什么需要分布式进程？
1.1 单机多进程的瓶颈
假设你有一台 8 核 CPU、16GB 内存的服务器：
| 资源 | 上限 | 瓶颈表现 |
|---|---|---|
| CPU | 8 核 | 计算任务排队，响应变慢 |
| 内存 | 16GB | 进程太多，内存不足，频繁 swap |
| 网络 | 单网卡 | 大量请求打满带宽 |
| 磁盘 IO | 单硬盘 | 读写速度成为瓶颈 |
无论怎么优化代码，硬件资源是硬约束。
1.2 垂直扩展 vs 水平扩展
| 方案 | 做法 | 优点 | 缺点 |
|---|---|---|---|
| 垂直扩展 | 换更强的机器（32核、128GB） | 简单，不用改代码 | 成本高，有上限，单点故障 |
| 水平扩展 | 加机器，分布式部署 | 成本低，几乎无上限，高可用 | 需要改架构，网络通信开销 |
分布式进程就是水平扩展——用多台普通机器组成集群，替代一台超级机器。
1.3 生活化理解
想象双 11 的快递仓库：
- 垂直扩展：把仓库建得更大、买更多叉车、雇更多工人——但仓库再大也有上限；
- 水平扩展：在全国建 100 个分仓，总仓统一调度，就近发货——可以无限扩张。
分布式进程就是"建分仓"的思路。
二、分布式进程的核心：网络化的 Queue
2.1 单机多进程的 Queue
在单机多进程中，我们用 multiprocessing.Queue 实现进程间通信：
python
from multiprocessing import Process, Queue
def worker(q):
while True:
task = q.get()
if task is None:
break
print(f'处理任务: {task}')
if __name__ == '__main__':
q = Queue()
p = Process(target=worker, args=(q,))
p.start()
for i in range(10):
q.put(f'任务{i}')
q.put(None) # 结束信号
p.join()Queue 是内存中的队列，只能在一台机器的进程间共享。
2.2 分布式进程的 Queue
分布式进程的关键创新：把 Queue 暴露到网络上，让其他机器的进程也能访问。
┌─────────────────────┐ Network ┌─────────────────────┐
│ 机器 A (Master) │ ◄────────────────────► │ 机器 B (Worker) │
│ │ TCP/IP 网络通信 │ │
│ ┌─────────────┐ │ │ ┌─────────────┐ │
│ │ task_queue │◄──┼──── get_task_queue() ────┼──┤ 获取任务 │ │
│ │ result_queue│◄──┼──── get_result_queue() ──┼──┤ 返回结果 │ │
│ └─────────────┘ │ │ └─────────────┘ │
│ QueueManager │ │ QueueManager │
└─────────────────────┘ └─────────────────────┘- Queue 对象实际存储在 Master 机器的内存中；
- Worker 通过网络"远程调用"Queue 的
get()和put()方法； - 这一切由
QueueManager自动完成，开发者无感知。
三、动手实战：搭建分布式任务系统
3.1 场景设定
我们要实现一个简单的分布式计算系统：
- Master（总店）：生成 10 个计算任务（计算 n²），放入任务队列；
- Worker（分店）：从任务队列取任务，计算结果，放入结果队列；
- Master 从结果队列读取最终结果。
Master 和 Worker 可以在同一台机器（测试用），也可以在不同机器（生产用）。
3.2 Master 端代码（task_master.py）
python
# task_master.py
import random
import queue
from multiprocessing.managers import BaseManager
# 发送任务的队列
task_queue = queue.Queue()
# 接收结果的队列
result_queue = queue.Queue()
# 从 BaseManager 继承的 QueueManager
class QueueManager(BaseManager):
pass
# 把两个 Queue 注册到网络上，callable 参数关联了 Queue 对象
QueueManager.register('get_task_queue', callable=lambda: task_queue)
QueueManager.register('get_result_queue', callable=lambda: result_queue)
# 绑定端口 5000，设置验证码 'abc'
manager = QueueManager(address=('', 5000), authkey=b'abc')
# 启动 QueueManager
manager.start()
# 获得通过网络访问的 Queue 对象
task = manager.get_task_queue()
result = manager.get_result_queue()
# 放几个任务进去
for i in range(10):
n = random.randint(0, 10000)
print('Put task %d...' % n)
task.put(n)
# 从 result 队列读取结果
print('Try get results...')
for i in range(10):
r = result.get(timeout=10)
print('Result: %s' % r)
# 关闭
manager.shutdown()
print('master exit.')3.3 Worker 端代码（task_worker.py）
python
# task_worker.py
import time
import queue
from multiprocessing.managers import BaseManager
# 创建类似的 QueueManager
class QueueManager(BaseManager):
pass
# 由于这个 QueueManager 只从网络上获取 Queue，所以注册时只提供名字
QueueManager.register('get_task_queue')
QueueManager.register('get_result_queue')
# 连接到服务器，也就是运行 task_master.py 的机器
server_addr = '127.0.0.1' # 如果是远程机器，改成 Master 的 IP
print('Connect to server %s...' % server_addr)
# 端口和验证码注意保持与 task_master.py 设置的完全一致
m = QueueManager(address=(server_addr, 5000), authkey=b'abc')
# 从网络连接
m.connect()
# 获取 Queue 的对象
task = m.get_task_queue()
result = m.get_result_queue()
# 从 task 队列取任务，并把结果写入 result 队列
for i in range(10):
try:
n = task.get(timeout=1)
print('run task %d * %d...' % (n, n))
r = '%d * %d = %d' % (n, n, n * n)
time.sleep(1) # 模拟耗时计算
result.put(r)
except queue.Empty:
print('task queue is empty.')
# 处理结束
print('worker exit.')3.4 运行效果
先启动 Master：
bash
$ python task_master.py
Put task 3411...
Put task 1605...
Put task 1398...
...
Try get results...再启动 Worker（同一台机器或另一台机器）：
bash
$ python task_worker.py
Connect to server 127.0.0.1...
run task 3411 * 3411...
run task 1605 * 1605...
...
worker exit.Master 输出结果：
Result: 3411 * 3411 = 11634921
Result: 1605 * 1605 = 2576025
...
master exit.四、代码深度解析
4.1 QueueManager 的工作原理
┌─────────────────────────────────────────┐
│ task_master.py │
│ │
│ task = manager.get_task_queue() │
│ result = manager.get_result_queue() │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────┐ │
│ │ QueueManager │ │
│ │ ┌────────────┐ ┌────────────┐ │ │
│ │ │ task_queue │ │result_queue│ │ │
│ │ └────────────┘ └────────────┘ │ │
│ └─────────────────────────────────┘ │
│ ▲ │
│ │ 网络代理 │
└──────────────┼──────────────────────────┘
│
Network
│
┌──────────────┼──────────────────────────┐
│ │ │
│ ┌───────────┴───────────┐ │
│ │ QueueManager │ │
│ │ （网络代理客户端） │ │
│ └────────────────────────┘ │
│ task_worker.py │
└─────────────────────────────────────────┘关键点：
- Queue 存储在 Master：
task_worker.py中没有创建 Queue 的代码，Queue 对象只在 Master 的内存中； - Worker 通过代理访问：Worker 拿到的是 Queue 的"网络代理"，所有操作都通过网络转发到 Master；
- QueueManager 是桥梁：负责序列化数据、网络传输、反序列化。
4.2 register() 方法的两个角色
| 角色 | 注册方式 | 说明 |
|---|---|---|
| Master | register('get_task_queue', callable=lambda: q) | 提供 callable，返回真实的 Queue 对象 |
| Worker | register('get_task_queue') | 只提供名字，用于网络查找 |
Master 端的 callable 参数是关键——当 Worker 通过网络请求 get_task_queue 时，Master 执行这个 callable，返回真实的 Queue。
4.3 authkey 的作用
python
manager = QueueManager(address=('', 5000), authkey=b'abc')authkey 是身份验证密钥，防止未授权的机器接入：
- Master 和 Worker 必须使用相同的
authkey才能连接； - 如果网络环境不可信，应该使用更复杂的密钥；
- 生产环境建议用 SSL/TLS 加密（
managers支持 SSL，但配置较复杂）。
4.4 为什么不能直接用原始 Queue？
python
# ❌ 错误：直接操作原始 Queue，绕过 QueueManager
task_queue.put(n)
# ✅ 正确：通过 manager.get_task_queue() 获取网络代理
task = manager.get_task_queue()
task.put(n)直接操作原始 Queue 只在 Master 本地有效，Worker 根本访问不到。必须通过 manager.get_xxx() 获取的代理对象，才能实现网络通信。
五、分布式进程的典型应用场景
5.1 邮件/短信队列
python
# Master：接收发送请求，放入队列
def on_user_register(email):
task.put({'type': 'send_email', 'to': email, 'template': 'welcome'})
# Worker：从队列取任务，真正发送
def worker():
while True:
task = task.get()
if task['type'] == 'send_email':
send_email(task['to'], task['template'])好处：注册接口秒回，邮件异步发送，不阻塞用户。
5.2 图片/视频处理
python
# Master：接收上传，分发处理任务
def on_upload(image_path):
task.put({'type': 'resize', 'path': image_path, 'sizes': [800, 400, 200]})
# Worker：多台机器并行处理图片
def worker():
while True:
task = task.get()
if task['type'] == 'resize':
resize_image(task['path'], task['sizes'])好处：上传后立即可用缩略图，原图后台处理。
5.3 爬虫系统
python
# Master：分配 URL
def master():
for url in url_list:
task.put({'type': 'crawl', 'url': url})
# Worker：多台机器并行爬取
def worker():
while True:
task = task.get()
if task['type'] == 'crawl':
html = fetch(task['url'])
result.put({'url': task['url'], 'html': html})好处：突破单机 IP 限制和带宽限制。
5.4 分布式日志分析
python
# Master：分发日志文件路径
def master():
for log_file in log_files:
task.put({'type': 'analyze', 'path': log_file}) # 只传路径，不传内容！
# Worker：读取共享存储上的日志
def worker():
while True:
task = task.get()
if task['type'] == 'analyze':
# 从 NFS/S3 读取日志文件
content = read_from_shared_storage(task['path'])
analyze(content)关键原则：队列里只传"引用"（文件路径、数据库 ID），不传"内容"（几百 MB 的文件本身）。
六、常见误区解析
误区 1：Queue 里可以传任意大的数据
真相：网络传输有开销，Queue 里应该只传"小数据"。
| 错误做法 | 正确做法 |
|---|---|
task.put({'data': 几百MB的文件内容}) | task.put({'path': '/shared/file.log'}) |
Worker 拿到路径后，自己去共享存储（NFS、S3、数据库）读取。
误区 2：分布式进程就是多进程
真相：分布式进程是多进程的"网络版"，但两者解决的问题不同。
| 对比项 | 单机多进程 | 分布式进程 |
|---|---|---|
| 目标 | 利用单机多核 | 突破单机资源上限 |
| 通信方式 | 内存 Queue | 网络 Queue |
| 机器数量 | 1 台 | 多台 |
| 网络依赖 | 无 | 有 |
误区 3：分布式进程不需要考虑网络问题
真相：网络是不稳定的，必须考虑：
- 网络延迟：Worker 取任务有网络延迟，比内存 Queue 慢；
- 网络断开：Worker 断线后，已取走的任务可能丢失；
- 防火墙：Master 的端口必须对 Worker 开放；
- 带宽限制：大量小任务可能打满内网带宽。
误区 4：authkey 足够安全
真相：authkey 只是简单的共享密钥，不防中间人攻击。
生产环境建议：
- 使用 SSL/TLS 加密（
multiprocessing.managers支持，但配置复杂）； - 或者改用更成熟的方案（Celery、RabbitMQ、Redis Queue）。
误区 5：Worker 越多越快
真相：受限于 Master 的处理能力和网络带宽。
- Master 是单点，所有 Queue 操作都经过它；
- Worker 太多时，Master 成为瓶颈；
- 需要监控 Master 的 CPU 和网络使用率。
七、进阶：生产环境的替代方案
multiprocessing.managers 适合学习和简单场景，但生产环境有更成熟的方案：
| 方案 | 特点 | 适用场景 |
|---|---|---|
| Celery | 功能丰富，支持定时任务、结果后端、监控 | 大型分布式任务队列 |
| RabbitMQ | 专业消息队列，高可靠、路由灵活 | 企业级消息系统 |
| Redis Queue (RQ) | 基于 Redis，简单轻量 | 中小型项目 |
| Kafka | 高吞吐、持久化、流处理 | 大数据、日志收集 |
| Dask | 分布式数据科学计算 | 大规模数据处理 |
何时从 managers 迁移？
- 需要任务持久化（机器重启不丢任务）；
- 需要任务优先级、定时任务；
- 需要监控面板、失败重试；
- 需要跨语言支持（Python + Java + Go）。
八、小结
| 核心知识点 | 一句话总结 |
|---|---|
| 分布式进程 | 把任务分发到多台机器执行，突破单机瓶颈 |
| QueueManager | 把本地 Queue 暴露到网络，Worker 通过代理访问 |
| Master 端 | 创建 Queue、注册 callable、启动 manager、放入任务 |
| Worker 端 | 注册同名方法、connect、获取代理、执行任务 |
| authkey | 简单的身份验证，防止未授权接入 |
| 数据原则 | Queue 只传"引用"（路径/ID），不传"内容" |
| 适用场景 | 邮件队列、图片处理、爬虫、日志分析 |
| 生产替代 | Celery、RabbitMQ、RQ 等更成熟的方案 |
记住三句话：
- 分布式进程 = 多进程 + 网络通信，
managers模块帮你封装了网络细节； - Queue 存在 Master 上，Worker 通过网络代理远程操作；
- 队列里只传"小数据"（路径、ID），大数据走共享存储。
参考代码
本文所有示例代码基于廖雪峰 Python 教程的"分布式进程"章节重新整理和扩充。