python - 如何在 Python 中创建有界内存装饰器？

Question

显然，快速搜索可以找到 Python 中内存装饰器的上百万种实现和风格。但是，我对一种我一直找不到的味道很感兴趣。我想让它的存储值的缓存可以是固定容量。添加新元素时，如果达到容量，则删除最旧的值并用最新值替换。

我担心的是，如果我使用 memoization 来存储大量元素，那么程序会因为内存不足而崩溃。 (我不知道这个问题在实践中有多重要。)如果缓存的大小是固定的，那么内存错误就不是问题。随着程序的执行，我处理的许多问题都会发生变化，因此初始缓存值与后来的缓存值看起来非常不同(并且以后出现的可能性要小得多)。这就是为什么我希望用最新的东西取代最旧的东西。

我找到了 OrderedDict 类和一个示例，该示例显示了如何对其进行子类化以指定最大大小。我想将其用作我的缓存，而不是普通的 dict。问题是，我需要 memoize 装饰器采用一个名为 maxlen 的参数，该参数默认为 None。如果它是 None，那么缓存是无限的并且正常运行。任何其他值都用作缓存的大小。

我希望它像下面这样工作:

@memoize
def some_function(spam, eggs):
    # This would use the boundless cache.
    pass

和

@memoize(200)  # or @memoize(maxlen=200)
def some_function(spam, eggs):
    # This would use the bounded cache of size 200.
    pass

下面是我到目前为止的代码，但我不知道如何将参数传递给装饰器，同时使其既“裸”又带有参数。

import collections
import functools

class BoundedOrderedDict(collections.OrderedDict):
    def __init__(self, *args, **kwds):
        self.maxlen = kwds.pop("maxlen", None)
        collections.OrderedDict.__init__(self, *args, **kwds)
        self._checklen()

    def __setitem__(self, key, value):
        collections.OrderedDict.__setitem__(self, key, value)
        self._checklen()

    def _checklen(self):
        if self.maxlen is not None:
            while len(self) > self.maxlen:
                self.popitem(last=False)

def memoize(function):
    cache = BoundedOrderedDict()  # I want this to take maxlen as an argument
    @functools.wraps(function)
    def memo_target(*args):
        lookup_value = args
        if lookup_value not in cache:
            cache[lookup_value] = function(*args)
        return cache[lookup_value]
    return memo_target

@memoize
def fib(n):
    if n < 2: return 1
    return fib(n-1) + fib(n-2)

if __name__ == '__main__':
    x = fib(50)
    print(x)

编辑:根据 Ben 的建议，我创建了以下装饰器，我相信它会按我想象的方式工作。能够将这些修饰函数与 multiprocessing 一起使用对我来说很重要，这在过去一直是个问题。但对该代码的快速测试似乎可以正常工作，即使将作业分包给线程池也是如此。

def memoize(func=None, maxlen=None):
    if func:
        cache = BoundedOrderedDict(maxlen=maxlen)
        @functools.wraps(func)
        def memo_target(*args):
            lookup_value = args
            if lookup_value not in cache:
                cache[lookup_value] = func(*args)
            return cache[lookup_value]
        return memo_target
    else:
        def memoize_factory(func):
            return memoize(func, maxlen=maxlen)
        return memoize_factory

Answer 1

@memoize
def some_function(spam, eggs):
    # This would use the boundless cache.
    pass

这里 memoize 被用作在单个函数参数上调用的函数，并返回一个函数。 memoize 是一个装饰器。

@memoize(200)  # or @memoize(maxlen=200)
def some_function(spam, eggs):
    # This would use the bounded cache of size 200.
    pass

这里 memoize 被用作一个函数，在单个整数参数上调用并返回一个函数，返回的函数本身用作装饰器，即它在单个函数参数上调用，并且返回一个函数。 memoize 是一个装饰器工厂。

所以要统一这两者，您将不得不编写一些丑陋的代码。我可能会这样做的方式是让 memoize 看起来像这样:

def memoize(func=None, maxlen=None):
    if func:
        # act as decorator
    else:
        # act as decorator factory

这样，如果你想传递参数，你总是将它们作为关键字参数传递，留下func(应该是一个位置参数)未设置，如果你只是想一切都默认它会神奇地直接作为装饰器工作。这确实意味着 @memoize(200) 会给你一个错误；您可以通过进行一些类型检查以查看 func 是否可调用来避免这种情况，这在实践中应该运行良好但并不是真正的“pythonic”。

另一种方法是使用两个不同的装饰器，比如 memoize 和 bounded_memoize。无界的 memoize 可以有一个简单的实现，只需调用 bounded_memoize 并将 maxlen 设置为 None，所以它不会在实现或维护方面不会花费您任何费用。

通常作为经验法则，我尽量避免破坏一个函数来实现两个仅切线相关的功能集，尤其是当它们具有如此不同的签名时。但在这种情况下，它确实使装饰器的使用变得很自然(要求@memoize()会很容易出错，尽管从理论上讲它更一致) ，并且您可能会实现一次并多次使用它，因此使用时的可读性可能是更重要的问题。