如何在Redis中实现元素可单独30天过期的类Python集合？

数据类型选择 #

Redis原生Set不支持单个元素设置TTL，Hash的TTL仅作用于整个键（无法实现元素独立过期），因此最优方案是将每个集合元素作为独立的Redis键，值设为占位符（如1），同时为每个键设置30天的TTL。该方案满足：

• 成员检查时间复杂度O(1)（通过EXISTS命令）
• 单个元素独立过期，无需手动维护清理
• 批量操作可通过管道高效完成

对应aioredis API说明 #

• 初始化：使用aioredis.from_url创建Redis客户端连接
• 批量添加：通过Redis管道（Pipeline）批量执行SET命令，附带EX参数设置30天（2592000秒）TTL，可选NX参数避免覆盖已存在的元素
• 成员检查：使用EXISTS命令判断元素对应的键是否存在

具体实现代码 #

先调整抽象类适配aioredis的异步特性，再完成具体逻辑：

from abc import abstractmethod, ABC
from typing import Iterable
import aioredis

class RedisSet(ABC):
    """实现基于Redis的字符串集合，支持元素级TTL"""
    @abstractmethod
    async def __init__(self, url: str):
        """
        初始化Redis连接，初始集合为空
        """
        raise NotImplementedError
    
    @abstractmethod
    async def add(self, new_elements: Iterable[str]) -> None:
        """将所有元素插入集合，每个元素30天后自动过期"""
        raise NotImplementedError
    
    @abstractmethod
    async def __contains__(self, elem: str) -> bool:
        """检查元素是否在集合中"""
        raise NotImplementedError

class RedisTTLSet(RedisSet):
    def __init__(self, url: str):
        self.redis = aioredis.from_url(url)
        # 30天对应的秒数
        self.ttl_seconds = 30 * 24 * 60 * 60

    async def add(self, new_elements: Iterable[str]) -> None:
        async with self.redis.pipeline(transaction=False) as pipe:
            for elem in new_elements:
                # SET key value EX ttl NX（可选NX避免覆盖已存在元素，去掉则会刷新TTL）
                await pipe.set(elem, 1, ex=self.ttl_seconds, nx=True)
            await pipe.execute()

    async def __contains__(self, elem: str) -> bool:
        result = await self.redis.exists(elem)
        return result == 1

注意事项 #

• 建议给元素键添加统一前缀（如my_set:{elem}），避免与其他Redis键冲突
• 若需要重复插入元素时刷新其过期时间，去掉SET命令中的NX参数即可
• 若使用Redis集群，统一前缀可保证键的哈希槽分布符合集群规则

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