Python 3.6是否有Erlang风格消息队列或多进程可用的共享内存实现？

核心实现思路 #

你需要的普通消息FIFO、高优先级消息优先出队的逻辑不需要完全复用通用队列的出队规则，用双队列拆分存储+优先级读取的方案就可以实现，完全可以避开multiprocessing.Manager的额外开销。

标准库无依赖实现方案（推荐） #

直接使用Python 3.6原生支持的multiprocessing.Queue即可，该队列基于管道+进程锁实现，本身多生产者安全，不需要启动额外管理进程，也没有TCP通信开销。
你可以拆分两个独立队列分别存储普通消息和高优先级（系统）消息，消费者每次优先读取高优先级队列即可满足需求：

import multiprocessing as mp
from typing import Any

def producer(common_queue: mp.Queue, sys_queue: mp.Queue, is_sys_msg: bool, msg: Any):
    """生产者逻辑，按消息类型写入对应队列"""
    if is_sys_msg:
        sys_queue.put(msg)
    else:
        common_queue.put(msg)

def consumer(common_queue: mp.Queue, sys_queue: mp.Queue):
    """消费者逻辑，优先取系统消息"""
    while True:
        # 优先处理系统消息
        if not sys_queue.empty():
            msg = sys_queue.get()
            # 此处添加系统消息处理逻辑
            print(f"收到系统消息: {msg}")
            continue
        # 无系统消息时处理普通消息
        msg = common_queue.get()
        # 此处添加普通消息处理逻辑
        print(f"收到普通消息: {msg}")
        # 自定义退出条件，可根据业务调整
        if msg == "exit":
            break

if __name__ == "__main__":
    # 初始化两个独立队列，分别存普通消息和系统消息
    common_q = mp.Queue()
    sys_q = mp.Queue()

    # 启动多个生产者
    p1 = mp.Process(target=producer, args=(common_q, sys_q, False, "普通消息1"))
    p2 = mp.Process(target=producer, args=(common_q, sys_q, True, "系统消息1"))
    p3 = mp.Process(target=producer, args=(common_q, sys_q, False, "普通消息2"))
    p1.start()
    p2.start()
    p3.start()

    # 启动消费者
    c = mp.Process(target=consumer, args=(common_q, sys_q))
    c.start()

    # 等待生产者执行完成，传入退出指令关闭消费者
    p1.join()
    p2.join()
    p3.join()
    common_q.put("exit")
    c.join()

如果有超过2个优先级的需求，按优先级等级初始化对应数量的队列，消费者从高到低遍历队列取非空数据即可。

更高性能的共享内存实现方案 #

如果对吞吐量要求极高，不想使用管道的额外开销，可以基于标准库multiprocessing.sharedctypes自己实现共享内存缓冲区：

• 用sharedctypes.Array分配固定大小的共享内存块，按优先级拆分不同的存储区域，或者分别初始化多个内存块对应不同优先级队列
• 用sharedctypes.Value存储每个队列的读指针、写指针、元素计数
• 全局使用一把multiprocessing.Lock保障所有读写操作的原子性，避免竞态问题
• 消息序列化可以用原生pickle，或者更轻量的第三方库msgpack（Python 3.6完全兼容）

如果你不想自己实现缓冲区逻辑，也可以使用第三方库posix-ipc，该库封装了POSIX标准的共享内存、信号量接口，完全支持Python 3.6，比标准库的共享内存接口更易用，同样不需要启动额外进程。

内容的提问来源于stack exchange，提问作者Incömplete