Python实现无TCP、无Redis的带过期时间键值存储方案咨询

当然可以！完全不用Redis或者TCP协议，咱们纯Python就能实现类似Nginx Shared Dict的本地键值存储，还自带自动过期清理功能。核心思路是用内存字典存储带过期时间戳的键值对，配合线程锁保证线程安全，再通过懒清理或后台定时清理来自动移除过期条目。下面给你一个完整的可运行实现：

核心设计思路 #

• 用Python内置字典作为底层存储，每个键对应的值是一个元组：(实际存储值, 过期时间戳)
• 加入threading.Lock确保多线程环境下的读写安全（毕竟Python字典不是线程安全的）
• 过期清理提供两种方案：
  • 懒清理：每次执行get_store或set_store时，顺便清理所有过期条目（适合低访问量场景，无额外线程开销）
  • 定时主动清理：启动一个后台线程，每隔固定时间扫描并清理过期条目（适合高访问量场景，避免每次读写都遍历字典）

完整实现代码 #

import threading
import time

class LocalKVStore:
    def __init__(self, clean_interval=None):
        # 底层存储字典：key -> (value, expire_timestamp)
        self._store = {}
        # 线程锁，保证多线程安全
        self._lock = threading.Lock()
        # 定时清理线程（如果指定了清理间隔）
        self._clean_thread = None
        self._stop_clean = threading.Event()
        
        if clean_interval is not None and clean_interval > 0:
            # 启动定时清理线程
            self._start_clean_thread(clean_interval)
    
    def _start_clean_thread(self, interval):
        def clean_task():
            while not self._stop_clean.is_set():
                with self._lock:
                    # 遍历并删除所有过期条目
                    current_time = time.time()
                    expired_keys = [k for k, (_, exp) in self._store.items() if exp <= current_time]
                    for k in expired_keys:
                        del self._store[k]
                # 等待指定间隔
                time.sleep(interval)
        
        self._clean_thread = threading.Thread(target=clean_task, daemon=True)
        self._clean_thread.start()
    
    def set_store(self, key, value, expire=None):
        """
        设置键值对
        :param key: 存储的键
        :param value: 存储的值
        :param expire: 过期时间（秒），None表示永不过期
        """
        current_time = time.time()
        # 计算过期时间戳：如果expire为None，设置为无穷大
        expire_timestamp = current_time + expire if expire is not None else float('inf')
        
        with self._lock:
            # 懒清理：设置前先清理过期条目（可选，根据需求开启）
            # self._clean_expired()
            self._store[key] = (value, expire_timestamp)
    
    def get_store(self, key):
        """
        获取键值对，过期则返回None
        :param key: 要获取的键
        :return: 存储的值，或None（键不存在/已过期）
        """
        current_time = time.time()
        
        with self._lock:
            # 先检查键是否存在
            if key not in self._store:
                return None
            
            value, expire_timestamp = self._store[key]
            # 检查是否过期
            if expire_timestamp <= current_time:
                # 过期则删除并返回None
                del self._store[key]
                return None
            
            # 懒清理：获取时顺便清理所有过期条目（可选）
            # self._clean_expired()
            return value
    
    def _clean_expired(self):
        """内部方法：清理所有过期条目"""
        current_time = time.time()
        expired_keys = [k for k, (_, exp) in self._store.items() if exp <= current_time]
        for k in expired_keys:
            del self._store[k]
    
    def __del__(self):
        # 停止定时清理线程
        if self._clean_thread is not None:
            self._stop_clean.set()
            self._clean_thread.join()

使用示例 #

# 初始化存储，可选开启定时清理（比如每30秒清理一次）
kv_store = LocalKVStore(clean_interval=30)

# 设置一个10秒后过期的键值对
kv_store.set_store("username", "sweetngx", expire=10)

# 立即获取，能拿到值
print(kv_store.get_store("username"))  # 输出: sweetngx

# 等待11秒后再获取
time.sleep(11)
print(kv_store.get_store("username"))  # 输出: None

注意事项 #

• 线程安全：代码中加入了threading.Lock，可以放心在多线程环境（比如Web服务）中使用
• 清理策略选择：如果你的服务访问量较低，推荐开启懒清理（取消set_store和get_store中的注释）；如果访问量高，建议用定时清理，避免每次读写都遍历字典带来的性能损耗
• 内存限制：因为是纯内存存储，数据量过大时会占用较多内存，你可以额外实现LRU淘汰逻辑（比如结合collections.OrderedDict）来限制最大存储数量
• 持久化扩展：如果需要重启服务后保留数据，可以添加定时将字典dump到本地文件的功能，初始化时再加载回来（这个属于可选扩展，根据你的需求来）

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