哈希算法题目：设计一个基于哈希的会话存储系统

题目描述
设计一个基于哈希的会话存储系统，用于管理用户会话数据。系统需要支持以下操作：

• createSession(sessionId, userData, ttl): 创建新会话，包含会话ID、用户数据和生存时间（TTL）
• getSession(sessionId): 根据会话ID获取用户数据，如果会话已过期返回空
• updateSession(sessionId, userData): 更新指定会话的用户数据
• deleteSession(sessionId): 删除指定会话
• cleanExpiredSessions(): 清理所有过期会话

解题过程

步骤1：分析需求和数据模型
会话存储系统需要：

1. 快速通过sessionId查找会话数据（O(1)时间复杂度）
2. 支持会话过期自动清理
3. 支持会话数据的增删改查

数据结构设计：

• 使用哈希表存储sessionId到会话数据的映射
• 需要额外数据结构管理会话过期时间

步骤2：设计会话数据结构

class Session:
    def __init__(self, session_id, user_data, ttl):
        self.session_id = session_id
        self.user_data = user_data
        self.created_time = time.time()  # 创建时间戳
        self.expire_time = self.created_time + ttl  # 过期时间戳
        self.ttl = ttl  # 生存时间（秒）

步骤3：基础哈希表实现

class SessionStorage:
    def __init__(self):
        self.sessions = {}  # 哈希表：session_id -> Session对象
        self.expiration_queue = []  # 用于管理过期的辅助结构

步骤4：实现createSession方法

def createSession(self, session_id, user_data, ttl):
    # 检查会话ID是否已存在
    if session_id in self.sessions:
        raise ValueError("Session ID already exists")
    
    # 创建新会话对象
    new_session = Session(session_id, user_data, ttl)
    
    # 存储到哈希表
    self.sessions[session_id] = new_session
    
    # 添加到过期队列（按过期时间排序）
    self._add_to_expiration_queue(new_session)

步骤5：实现_getSession辅助方法（包含过期检查）

def _getSession(self, session_id):
    if session_id not in self.sessions:
        return None
    
    session = self.sessions[session_id]
    
    # 检查会话是否过期
    if time.time() > session.expire_time:
        self._removeSession(session_id)
        return None
    
    return session

步骤6：实现getSession方法

def getSession(self, session_id):
    session = self._getSession(session_id)
    return session.user_data if session else None

步骤7：实现updateSession方法

def updateSession(self, session_id, user_data):
    session = self._getSession(session_id)
    if session is None:
        raise ValueError("Session not found or expired")
    
    # 更新用户数据
    session.user_data = user_data
    
    # 可选：更新过期时间（根据需求决定）
    # session.expire_time = time.time() + session.ttl

步骤8：实现deleteSession方法

def deleteSession(self, session_id):
    self._removeSession(session_id)

def _removeSession(self, session_id):
    if session_id in self.sessions:
        # 从哈希表删除
        del self.sessions[session_id]
        # 从过期队列删除
        self._remove_from_expiration_queue(session_id)

步骤9：设计过期会话清理策略
方案1：惰性清理（访问时检查）

# 已经在_getSession中实现，访问时检查并清理

方案2：定期批量清理

def cleanExpiredSessions(self):
    current_time = time.time()
    sessions_to_remove = []
    
    # 找出所有过期会话
    for session_id, session in self.sessions.items():
        if current_time > session.expire_time:
            sessions_to_remove.append(session_id)
    
    # 批量删除过期会话
    for session_id in sessions_to_remove:
        self._removeSession(session_id)

步骤10：优化过期队列实现
使用最小堆优化过期时间管理：

import heapq

def __init__(self):
    self.sessions = {}
    self.expiration_heap = []  # 最小堆：(expire_time, session_id)

def _add_to_expiration_queue(self, session):
    heapq.heappush(self.expiration_heap, (session.expire_time, session.session_id))

def _remove_from_expiration_queue(self, session_id):
    # 惰性删除：在清理时跳过已删除的会话
    pass

def cleanExpiredSessions(self):
    current_time = time.time()
    
    while self.expiration_heap and self.expiration_heap[0][0] <= current_time:
        expire_time, session_id = heapq.heappop(self.expiration_heap)
        
        # 检查会话是否仍然存在且确实过期
        if (session_id in self.sessions and 
            self.sessions[session_id].expire_time <= current_time):
            del self.sessions[session_id]

步骤11：完整实现代码

import time
import heapq

class Session:
    def __init__(self, session_id, user_data, ttl):
        self.session_id = session_id
        self.user_data = user_data
        self.created_time = time.time()
        self.expire_time = self.created_time + ttl
        self.ttl = ttl

class SessionStorage:
    def __init__(self):
        self.sessions = {}
        self.expiration_heap = []
    
    def createSession(self, session_id, user_data, ttl):
        if session_id in self.sessions:
            raise ValueError("Session ID already exists")
        
        new_session = Session(session_id, user_data, ttl)
        self.sessions[session_id] = new_session
        heapq.heappush(self.expiration_heap, (new_session.expire_time, session_id))
    
    def getSession(self, session_id):
        session = self._getSession(session_id)
        return session.user_data if session else None
    
    def updateSession(self, session_id, user_data):
        session = self._getSession(session_id)
        if session is None:
            raise ValueError("Session not found or expired")
        session.user_data = user_data
    
    def deleteSession(self, session_id):
        if session_id in self.sessions:
            del self.sessions[session_id]
    
    def cleanExpiredSessions(self):
        current_time = time.time()
        
        while self.expiration_heap and self.expiration_heap[0][0] <= current_time:
            expire_time, session_id = heapq.heappop(self.expiration_heap)
            
            if (session_id in self.sessions and 
                self.sessions[session_id].expire_time <= current_time):
                del self.sessions[session_id]
    
    def _getSession(self, session_id):
        if session_id not in self.sessions:
            return None
        
        session = self.sessions[session_id]
        if time.time() > session.expire_time:
            self.deleteSession(session_id)
            return None
        
        return session

步骤12：复杂度分析

• 创建会话：O(log n) - 堆插入操作
• 查询会话：O(1) - 哈希表查找
• 更新会话：O(1) - 哈希表查找+数据修改
• 删除会话：O(1) - 哈希表删除（惰性堆清理）
• 清理过期会话：O(k log n) - k个过期会话的堆操作

这个设计结合了哈希表的快速查找和最小堆的高效过期管理，适合实际的会话存储场景。