在 Python 中实现多进程间共享缓存，并且确保写入操作不会阻塞读取操作，同时实现逐条处理数据，可以使用以下几种机制结合操作系统级别的共享内存和队列机制。 ### 使用 `multiprocessing.Manager` 实现共享缓存 Python 提供了 `multiprocessing.Manager` 用于在多个进程之间共享 Python 对象（如列表、字典等）。虽然性能不如底层共享内存，但适合用于实现进程间共享缓存。 ```python from multiprocessing import Process, Manager import time def writer(shared_list): for i in range(5): shared_list.append(i) time.sleep(0.1) # 模拟逐条写入 def reader(shared_list): while True: if shared_list: print("Read:", shared_list.pop(0)) else: time.sleep(0.1) if __name__ == '__main__': with Manager() as manager: shared_list = manager.list() p_writer = Process(target=writer, args=(shared_list,)) p_reader = Process(target=reader, args=(shared_list,)) p_writer.start() p_reader.start() p_writer.join() p_reader.join() ``` 此方式利用了 `multiprocessing.Manager` 提供的同步机制，确保多个进程之间共享数据时的一致性。[^2] ### 使用 `multiprocessing.Queue` 实现非阻塞读写 为了实现写入操作不阻塞读取操作，可以使用 `multiprocessing.Queue`，它是一个线程和进程安全的队列结构，非常适合用于生产者-消费者模式。 ```python from multiprocessing import Process, Queue import time def writer(queue): for i in range(5): queue.put(i) time.sleep(0.1) def reader(queue): while True: try: item = queue.get(timeout=1) # 设置超时避免永久阻塞 print("Read:", item) except: break if __name__ == '__main__': queue = Queue() p_writer = Process(target=writer, args=(queue,)) p_reader = Process(target=reader, args=(queue,)) p_writer.start() p_reader.start() p_writer.join() p_reader.join() ``` 此方式利用了队列的先进先出特性，确保数据逐条处理，并且读取操作不会因为写入操作而被阻塞。[^2] ### 使用 `multiprocessing.shared_memory` 实现高性能共享缓存 对于更高性能的场景，可以使用 `multiprocessing.shared_memory` 模块，它允许在多个进程之间共享一块原始内存区域。 ```python from multiprocessing import shared_memory, Process import numpy as np import time def writer(shm_name, size): existing_shm = shared_memory.SharedMemory(name=shm_name) np_array = np.ndarray((size,), dtype=np.int64, buffer=existing_shm.buf) for i in range(size): np_array[i] = i time.sleep(0.1) def reader(shm_name, size): existing_shm = shared_memory.SharedMemory(name=shm_name) np_array = np.ndarray((size,), dtype=np.int64, buffer=existing_shm.buf) while True: for i in range(size): if np_array[i] != -1: print("Read:", np_array[i]) np_array[i] = -1 # 标记为已读 if __name__ == '__main__': shm = shared_memory.SharedMemory(create=True, size=100) try: size = 10 # 初始化共享内存 np_array = np.ndarray((size,), dtype=np.int64, buffer=shm.buf) np_array.fill(-1) p_writer = Process(target=writer, args=(shm.name, size)) p_reader = Process(target=reader, args=(shm.name, size)) p_writer.start() p_reader.start() p_writer.join() p_reader.join() finally: shm.close() shm.unlink() ``` 此方式直接操作共享内存，适合需要高性能的场景。需要注意的是，共享内存的同步需要额外的机制（如锁或信号量）来保证数据一致性。[^4] ###