Python multiprocessing：需非消费式共享最新值的类Queue对象

完全同意你的思路——用共享内存方案确实比Queue/Pipe更合适！

你要的不是Queue那种FIFO消息队列（会保留所有历史值，get操作消耗元素），而是一个能让多进程共享、始终暴露最新写入值、读取不消耗值的"状态容器"。共享内存刚好完美匹配这个场景：它直接在进程间开辟一块共享的内存区域，更新时直接覆盖旧值，读取时直接获取当前值，不需要维护队列结构，效率更高，也完全满足你"旧值可直接丢弃"的需求。

具体实现方案 #

Python的multiprocessing模块提供了Value和Array两个共享内存工具，其中Value适合单个值的场景，正好对应你的需求。下面是一个符合你示例的实现：

from multiprocessing import Value, Process

class SharedLatestValue:
    def __init__(self, initial_value=None):
        # 根据存储的数据类型选择类型码：
        # 'i'=整数, 'd'=浮点数, 's'=字符串（需指定长度）, 'object'=任意可pickle对象
        self._shared_val = Value('i', initial_value if initial_value is not None else 0)
    
    def put(self, new_value):
        # 显式加锁保证并发写入安全（Value本身的value属性访问也会自动加锁，显式加锁更清晰）
        with self._shared_val.get_lock():
            self._shared_val.value = new_value
    
    def get(self):
        with self._shared_val.get_lock():
            return self._shared_val.value

# 测试你的示例场景
if __name__ == "__main__":
    q = SharedLatestValue()
    q.put(0)
    q.put(1)
    print(q.get())  # 输出1
    print(q.get())  # 仍然输出1
    q.put(2)
    print(q.get())  # 输出2

    # 多进程测试
    def worker(shared_obj):
        print(f"Worker读取值：{shared_obj.get()}")  # 输出2
        shared_obj.put(3)
        print(f"Worker更新后读取：{shared_obj.get()}")  # 输出3

    p = Process(target=worker, args=(q,))
    p.start()
    p.join()

    print(f"主进程最终读取：{q.get()}")  # 输出3

关键说明 #

1. 进程安全：Value自带互斥锁，通过get_lock()获取锁后进行读写操作，避免多进程并发读写导致的数据不一致问题。
2. 类型灵活性：如果需要存储字符串或自定义对象，可以改用Value('object', initial_value)，但要注意自定义对象必须是可pickle序列化的（Python大部分内置类型都满足）。
3. 对比Queue：Queue的设计目标是"传递消息"，每个get操作会移除队列中的元素；而这个共享容器的目标是"共享状态"，所有进程读取的都是当前最新的状态，完全符合你的需求。

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