在Redis中获得一个单独元素在30天内过期的集合的最干净的方法?

mwkjh3gx  于 2023-11-16  发布在  Redis
关注(0)|答案(1)|浏览(111)

我需要一个类似于Redis中Python set的数据结构,并具有在插入后30天自动从set中过期(弹出)的单个元素的额外功能。基本上,这些是我想要的类的抽象行为。

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有一个set数据类型,但似乎(如果我错了,很高兴得到纠正)它不支持任何生存时间(TTL)。Au contraire,字典支持TTL,但我必须为每个键使用一个占位符value(不必要的开销),我不确定成员检查是否将是常数时间。
N. B。

  • 我不认为有任何必要对这些元素进行遍历,所以对这些操作进行优化是多余的,我所需要的只是成员检查。
  • 我打算使用aioredis作为python客户端。它应该支持必要的redis操作。

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

zte4gxcn

zte4gxcn1#

要在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,则表示您刚刚更新了该元素的得分。

检查是否存在

现在你可以通过这个命令检查一个密钥的存在性,注意,你需要验证这个元素没有过期,所以在你的代码中需要检查时间和分数。

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


(nil)表示该元素不存在

正在删除过时的元素

这是由schaduler运行的,用于从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

相关问题