如何向Python multiprocessing queue传值并读取最新数据且不清空队列

针对你需要让读取进程始终获取最新数据的需求，这里给几个适合新手的简单方案：

方案一：用multiprocessing.Manager共享字典（最直观） #

Manager可以创建跨进程安全的字典，生产者直接更新字典中的指定键，消费者每次读取这个键就能拿到最新值，完全不用处理队列的移除逻辑。

from multiprocessing import Process, Manager
import time

def producer(shared_dict):
    # 写入最新数据，可替换为循环持续更新
    shared_dict['latest_data'] = [42, None, 'hello']

if __name__ == '__main__':
    with Manager() as manager:
        shared_dict = manager.dict()
        p = Process(target=producer, args=(shared_dict,))
        p.start()
        time.sleep(0.1)  # 等待生产者完成写入
        
        # 两次读取都拿到最新值
        print(shared_dict['latest_data'])
        print(shared_dict['latest_data'])
        
        p.join()

方案二：用multiprocessing.Value共享单一值（适合简单结构） #

如果数据是单一对象或可序列化的简单结构，用Value更轻量，指定ctypes.py_object就能存储任意Python对象：

from multiprocessing import Process, Value
import time
import ctypes

def producer(shared_val):
    shared_val.value = [42, None, 'hello']
    # 循环更新示例：
    # for i in range(5):
    #     shared_val.value = [i, None, f'hello_{i}']
    #     time.sleep(0.5)

if __name__ == '__main__':
    shared_val = Value(ctypes.py_object)
    p = Process(target=producer, args=(shared_val,))
    p.start()
    time.sleep(0.1)
    
    print(shared_val.value)
    print(shared_val.value)
    
    p.join()

方案三：封装Queue实现“只保留最新数据”（如果坚持用Queue） #

如果一定要基于Queue实现，可以封装一个类，确保队列中永远只存最新数据，读取时自动丢弃旧数据：

from multiprocessing import Process, Queue
import time

class LatestDataQueue:
    def __init__(self):
        self.queue = Queue(maxsize=1)
    
    def put(self, data):
        # 队列满时先移除旧数据，再放入新数据
        try:
            self.queue.get(block=False)
        except:
            pass
        self.queue.put(data)
    
    def get(self):
        # 清空队列所有旧数据，返回最后一条（最新的）
        latest = None
        while True:
            try:
                latest = self.queue.get(block=False)
            except:
                break
        return latest

def producer(q):
    q.put([42, None, 'hello'])

if __name__ == '__main__':
    q = LatestDataQueue()
    p = Process(target=producer, args=(q,))
    p.start()
    time.sleep(0.1)
    
    print(q.get())
    print(q.get())
    
    p.join()

总结 #

对于新手来说，前两个方案（共享字典/Value）最容易理解和实现，不需要处理队列的复杂逻辑，直接通过共享变量读写就能满足“始终获取最新数据”的需求。

内容的提问来源于stack exchange，提问作者Paul K.