分布式进程

本地多进程只是在一台机器上启动多个进程；分布式进程则把任务和工作进程放到不同机器上，通过网络交换任务和结果。它适合把计算或任务消费能力扩展到多台主机，但也引入网络延迟、连接失败、重复执行、认证和部署等问题。

Python 标准库的 multiprocessing.managers 可以快速演示一个分布式任务队列：一台机器运行管理器（服务器），其他机器连接它，取得队列的代理对象。这个方案适合教学、小型内网工具和原型，不等同于生产级消息队列。

架构和消息流

生产者进程 ──网络──> Manager 服务器 ──共享队列代理──> 工作进程
生产者进程 <──网络── Manager 服务器 <──结果队列代理── 工作进程

服务器进程持有真正的 Queue；客户端拿到的是代理。客户端调用 put()、get() 时，参数和返回值会经过序列化并通过网络传输。

编写 Manager 服务器

from multiprocessing import Queue
from multiprocessing.managers import BaseManager


task_queue = Queue()
result_queue = Queue()


class QueueManager(BaseManager):
    pass


QueueManager.register("get_task_queue", callable=lambda: task_queue)
QueueManager.register("get_result_queue", callable=lambda: result_queue)


if __name__ == "__main__":
    manager = QueueManager(address=("0.0.0.0", 50000), authkey=b"change-me")
    server = manager.get_server()
    print("Manager 正在监听 50000 端口")
    server.serve_forever()

register() 暴露可被远程取得的对象，address 指定监听地址和端口，authkey 提供基础认证。不要把示例密钥用于真实系统，也不要直接把管理器端口暴露到不可信公网；应使用内网、防火墙、VPN 或专用消息系统保护通信。

编写客户端生产者

from multiprocessing.managers import BaseManager


class QueueManager(BaseManager):
    pass


QueueManager.register("get_task_queue")
QueueManager.register("get_result_queue")


if __name__ == "__main__":
    manager = QueueManager(address=("manager.example.internal", 50000), authkey=b"change-me")
    manager.connect()
    tasks = manager.get_task_queue()
    results = manager.get_result_queue()

    for value in range(10):
        tasks.put((value, value + 1))

    for _ in range(10):
        print("结果：", results.get(timeout=30))

客户端注册同名方法但不传 callable，再调用 connect() 连接服务器。Queue.get(timeout=...) 让网络或工作进程异常时能够返回控制权，而不是无限阻塞。

编写工作进程

from multiprocessing.managers import BaseManager
from queue import Empty


class QueueManager(BaseManager):
    pass


QueueManager.register("get_task_queue")
QueueManager.register("get_result_queue")


def worker() -> None:
    manager = QueueManager(address=("manager.example.internal", 50000), authkey=b"change-me")
    manager.connect()
    tasks = manager.get_task_queue()
    results = manager.get_result_queue()

    while True:
        try:
            left, right = tasks.get(timeout=10)
        except Empty:
            print("暂时没有任务，工作进程退出")
            return
        try:
            results.put(left + right)
        except Exception as error:
            print("任务失败：", error)


if __name__ == "__main__":
    worker()

实际部署时，通常会把“没有任务”“连接断开”“任务失败”分别处理，而不是把所有异常都当成正常结束。任务应带有唯一 ID、重试次数和超时时间，结果也应能幂等写入，避免客户端重试时产生重复副作用。

任务协议应该包含什么

演示中的 (left, right) 过于简单。真实任务通常使用结构明确的消息：

task = {
    "id": "task-20260913-001",
    "type": "add",
    "payload": {"left": 10, "right": 20},
    "attempt": 1,
}

• id 用于去重、追踪和关联结果；
• type 决定处理器，工作进程不应执行消息携带的任意代码；
• payload 只放完成任务所需的数据；
• attempt 或时间戳用于限制重试并诊断延迟。

对应结果至少应包含任务 ID、成功状态、输出或经过筛选的错误信息。协议应有版本，并在消费者不能识别新字段或新任务类型时明确拒绝，而不是静默产生错误结果。

分布式系统新增的成本

网络不是可靠内存

本地队列通常很快，但远程调用可能遇到延迟、断线、半连接和服务器重启。一次 put() 成功并不一定代表业务数据已经持久化。需要可靠投递时，应使用支持确认、持久化和重试的消息队列。

客户端超时也不代表服务器没有执行任务：响应可能只是在返回途中丢失。只要存在自动重试，同一任务就可能执行两次。因此扣款、发信、写数据库等有副作用的操作必须基于任务 ID 实现幂等，或由具备事务能力的系统保证唯一提交。

序列化和兼容性

代理调用的参数和返回值必须可序列化。避免发送打开的文件、套接字、锁和复杂的本地对象；传递简单字典、列表、字符串和数字更稳妥。不同版本客户端和服务器还应约定消息格式，避免类路径变化导致反序列化失败。

认证、授权和机密

authkey 只能提供有限的连接认证，不应被当作完整的 TLS、身份授权和审计方案。生产系统至少要限制网络访问、轮换凭据、避免日志打印密钥，并对任务内容进行校验。不要对不可信来源直接反序列化任意对象。

监控和优雅关闭

至少监控队列积压、任务耗时、成功率、重试次数、工作进程心跳和 Manager 连接错误。停止服务时应先停止接收新任务，再等待正在执行的任务完成，最后关闭连接；直接终止进程可能让任务停在结果未知的状态。

标准库 Manager 没有提供完整的持久化、消费确认、死信队列和集群高可用能力。需要这些能力时，应尽早迁移到专门系统，而不是在代理队列外层逐项补造。

什么时候使用什么方案

需求	更合适的选择
单机 CPU 密集计算	ProcessPoolExecutor
教学或小型内网 RPC/队列原型	BaseManager
需要持久化、确认、重试和监控的任务	专业消息队列或任务系统
高吞吐服务间调用	明确定义协议的 RPC/HTTP/gRPC 系统
多节点数据处理	根据数据规模选择批处理、流处理或分布式计算框架

分布式进程的重点不是“把 Process 放到另一台机器”，而是设计清晰的任务协议：谁提交、谁负责、失败如何重试、结果如何确认、重复执行是否安全，以及系统如何观测和停止。