如何在Python中创建多进程共享的有限大小缓存

问题核心 #

multiprocessing.Manager返回的DictProxy是跨进程的代理对象，无法直接子类化——实际的字典实例运行在Manager的后台进程中，本地仅持有代理引用。你之前尝试的LimitedSizeDict返回空列表，大概率是因为没有基于共享对象实现，导致每个进程持有独立副本而非真正共享。

以下是两种可行的实现方案：

方案1：封装DictProxy实现LRU缓存 #

通过本地封装类，结合Manager的共享字典、共享列表维护访问顺序，加锁保证多进程操作的原子性，实现类似lru_cache的淘汰逻辑。

from multiprocessing import Manager, Lock

class SharedLRUCache:
    def __init__(self, maxsize=128):
        self.manager = Manager()
        # 共享缓存字典
        self.cache = self.manager.dict()
        # 共享列表维护访问顺序（末尾为最近使用）
        self.access_order = self.manager.list()
        self.maxsize = maxsize
        # 多进程同步锁
        self.lock = Lock()

    def __getitem__(self, key):
        with self.lock:
            if key not in self.cache:
                raise KeyError(key)
            # 更新访问顺序：移到末尾标记为最近使用
            self.access_order.remove(key)
            self.access_order.append(key)
            return self.cache[key]

    def __setitem__(self, key, value):
        with self.lock:
            if key in self.cache:
                # 已存在则更新值和访问顺序
                self.cache[key] = value
                self.access_order.remove(key)
                self.access_order.append(key)
                return
            
            # 新条目：检查是否超出容量
            if len(self.cache) >= self.maxsize:
                # 删除最久未使用的条目（列表头部）
                oldest_key = self.access_order.pop(0)
                del self.cache[oldest_key]
            
            self.cache[key] = value
            self.access_order.append(key)

    def get(self, key, default=None):
        try:
            return self[key]
        except KeyError:
            return default

使用方式 #

主进程创建实例后，直接传递给子进程即可：

from multiprocessing import Process

def worker(cache):
    cache["key1"] = "value1"
    print(cache.get("key1"))

if __name__ == "__main__":
    cache = SharedLRUCache(maxsize=5)
    p = Process(target=worker, args=(cache,))
    p.start()
    p.join()
    print(cache.get("key1"))  # 输出value1

方案2：注册自定义有限大小字典到Manager #

自定义一个带大小限制的字典类，注册到Manager中，让Manager直接创建共享的实例（底层仍通过代理实现跨进程访问）。

from multiprocessing.managers import BaseManager, DictProxy
from collections import OrderedDict

class LimitedSizeLRUDict(OrderedDict):
    def __init__(self, maxsize=128, *args, **kwargs):
        self.maxsize = maxsize
        super().__init__(*args, **kwargs)

    def __setitem__(self, key, value):
        if key in self:
            # 已存在则删除后重新插入，更新访问顺序
            del self[key]
        super().__setitem__(key, value)
        
        # 超出容量则删除最久未使用的条目
        if len(self) > self.maxsize:
            self.popitem(last=False)

    def __getitem__(self, key):
        # 访问时更新顺序，标记为最近使用
        value = super().__getitem__(key)
        del self[key]
        super().__setitem__(key, value)
        return value

# 注册自定义类到Manager
class CacheManager(BaseManager):
    pass

CacheManager.register('LimitedSizeLRUDict', LimitedSizeLRUDict, proxytype=DictProxy)

### 使用方式
if __name__ == "__main__":
    with CacheManager() as manager:
        cache = manager.LimitedSizeLRUDict(maxsize=10)
        # 子进程直接使用该cache对象即可实现共享

关键注意事项 #

1. 多进程操作必须加锁（方案1显式加锁，方案2由Manager代理保证线程安全），避免竞态条件。
2. 不要直接在本地创建自定义字典再传给子进程——这会导致每个进程持有独立副本，无法实现共享。

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