Python多进程可控双向队列如何实现跨进程定向消息通信

原生multiprocessing.Queue本身是无路由的竞争消费模型，默认不区分消息的发送方和接收方，所有挂在同一个队列上的进程都会抢占消费权，单队列共享必然会出现自产自消、消息错投的问题，靠加字段筛选的方式不仅会破坏顺序，还会带来大量无效轮询开销，可落地的实现方案有两种，根据业务场景选即可。

方案1：进程专属收件队列（生产环境首选，易扩展、无性能损耗） #

这是工程上最通用的实现方式，从架构上规避消息错拿问题，不需要对消息内容做任何修改：

• 进程初始化阶段，为每个进程创建一个仅由该进程负责消费的专属收件队列，同时维护一份全局可访问的「进程唯一标识 -> 对应收件队列」的映射表
• 每个进程只对自己的专属收件队列调用get()方法拉取任务，从根源上避免读到自己发出的消息、抢拿其他进程任务的问题
• 进程需要向指定目标投递任务时，直接从映射表中找到目标进程对应的收件队列，调用该队列的put()方法写入任务即可

这个方案的优势非常明显：

• 完全保留队列原生FIFO特性：每个进程收件箱内的消息严格按照投递顺序被消费，不存在乱序问题
• 扩展成本极低：新增进程只需要创建对应专属队列、更新全局映射表即可，原有通信逻辑不需要做任何改动，支持任意数量的进程横向扩展
• 无额外性能开销：不需要给消息加冗余标识，也不会出现进程拿到非自身任务再回塞队列的无效操作，性能和原生双队列双向通信完全一致
• 兼容原生Queue的所有特性：支持多生产者同时向同一个收件队列写任务，也支持一个进程同时向任意多个目标进程投递任务

补充性能参考：仅在两个进程一对一通信的场景下，multiprocessing.Pipe的传输延迟比双Queue低30%左右，但Pipe是点对点通信模型，进程数超过2个之后，连接管理的复杂度会陡增，可维护性远不如分收件队列的方案。

方案2：单队列+信封结构（仅适配全局强顺序要求的特殊场景） #

如果业务要求所有消息必须严格按照全局生成的先后顺序被消费，不能拆分队列，可以用信封封装+本地缓存的方式实现，但不推荐进程数超过3个的场景使用：

• 所有投递到全局共享队列的消息统一封装为(target_id, job_payload)格式，明确标注目标接收进程的唯一标识
• 每个进程维护一个本地的临时缓存队列，每次拉取消息时优先检查本地缓存：
  • 如果缓存不为空，从缓存头部取消息判断目标ID，匹配自身ID则处理，不匹配则放回缓存尾部
  • 如果缓存为空，从全局共享队列拉取新消息，匹配自身ID则处理，不匹配则写入本地缓存尾部
• 循环上述逻辑直到拿到属于自己的消息

注意：这个方案本质是把消息错配的筛选成本从队列层转移到了业务层，进程数越多，消息在不同进程缓存间无效流转的开销越大，高并发场景下性能下降非常明显；同时因为缓存机制的存在，所谓的全局顺序保证会被削弱，非特殊需求不要用。

常见避坑 #

不要尝试靠“写入后暂时跳过get调用”的方式规避自产自消问题，多进程并发写入时，你完全无法预判队列中下一条消息的发送方是谁，这种写法在并发场景下必然会出现消息错拿，没有可靠性可言。

内容的提问来源于stack exchange，提问作者Abdelhadi Abdelhadi