在 Redis 中获取单个元素在 30 天后过期的集合的最简洁方法？

Question

我需要一个类似于 Redis 中的 Python 集的数据结构，并具有插入后 30 天自动从集合中过期（弹出）的单个元素的附加功能。基本上，这些是我想要从课堂上得到的抽象行为。

from abc import abstractmethod, ABC
from typing import Iterable 

class RedisSet(ABC):
    """Implement a set of strings with a remote redis host."""
    @abstractmethod
    def __init__(self, url:str):
        """
           Initialise the class at the remote redis url.
           The set of strings should be empty initially.            
        """
        raise NotImplementedError    
    @abstractmethod
    def add(self, new_elements:Iterable[str])->None:
        """Insert all the elements into the set."""
        raise NotImplementedError
    @abstractmethod
    def __contains__(self, elem:str)->bool:
        """Check membership."""
        raise NotImplementedError

那么实现这一目标的最干净的方法是什么？我不需要实现整个类，而是询问在 redis 中使用的正确数据类型和 API 是什么，因为我并不完全熟悉 redis 的全部功能。

我注意到 Redis 有一个固定的数据类型，但似乎（如果我错了，很高兴得到纠正）它不支持任何生存时间（TTL）。相反，字典支持 TTL，但我必须为每个键使用占位符

value

（不必要的开销），并且我不确定成员资格检查是否会是恒定时间。

N。 B.

• 我预计不需要迭代这些元素，因此针对该操作进行优化是多余的。我所需要的只是会员资格检查。
• 我打算使用

  aioredis

  作为Python客户端。它应该支持必要的redis操作。

任何有关我应该查找其文档的正确 Redis 数据类型的想法将不胜感激。

Answer 1

要在

Redis

中实现您想要的类似功能，您应该考虑到

Redis

存在限制，并且最终解决方案可能看起来很复杂。

有什么限制？

正如您正确提到的，

Redis

不提供 HASH、SETS 或任何其他数据类型的

TTL

功能。只需支持简单的Key/Value即可。

解决办法是什么？

您可以使用SortedSet。此数据类型会为您添加到其中的每个元素获取一个分数，在本例中这将是过期时间。 您需要通过代码维护排序集以从中删除过期元素。您可以使用调度程序来实现这部分。

添加新元素

ZADD my-set 1699631227694 elm1
(integer) 1
ZADD my-set 1699631228234 elm2 1699631228234 elm3
(integer) 2
ZADD my-set 1699631229123 elm1
(integer) 0

(integer) ${number}

-

${number}

确定您添加的元素数量，如果您得到 0 或小于您输入的元素，则意味着您刚刚更新了该元素的分数。

检查是否存在

现在你可以通过这个命令检查一个key是否存在，注意，你需要验证elememnt没有过期，所以在你的代码中需要检查时间和分数。

ZSCORE my-set elm1 // which retuen the score of elm1
"1699631227694"
ZSCORE my-set elm4
(nil)

(nil)

表示该元素不存在

删除过时的元素

这由您的调度程序运行，以删除从

sorted set

中过期的元素

ZREMRANGEBYSCORE my-set -inf 1699631228234
(integer) 2

不幸的是，我不是Python开发人员，无法帮助您编写代码。 但我认为这段代码可以帮助您获得洞察力：

from abc import abstractmethod, ABC
from typing import Iterable 

class RedisSet(ABC):
    """Implement a set of strings with a remote redis host."""
    @abstractmethod
    def __init__(self, url:str):
        """
           Initialise the class at the remote redis url.
           The set of strings should be empty initially.            
        """
        self.redis = // init the redis connection
        self.set_key = set_key
        
    @abstractmethod
    def add(self, new_elements:Iterable[str])->None:
        """Insert all the elements into the set."""
        const startTime = Date.now();
        const offset = 1000*60*60*24*30; // 30 days
        
        for elem in new_elements:
            await self.redis..zadd(self.set_key, Date.now() + offset, elem)

    @abstractmethod
    def __contains__(self, elem:str)->bool:
        """Check membership."""
        exists = await self.redis.zscore(self.set_key, elem)

        if score is None or score < time.time():
            return False
        
        return True