import threading import time import logging from typing import Dict, Callable, Any, Optional, List, Tuple from collections import OrderedDict logger = logging.getLogger(__name__) # ---------- 脏标记管理 ---------- _dirty_set = set() _dirty_lock = threading.Lock() def mark_dirty(name: str): with _dirty_lock: _dirty_set.add(name) def clear_dirty(name: str): with _dirty_lock: _dirty_set.discard(name) def get_dirty_list() -> List[str]: with _dirty_lock: return list(_dirty_set) class PlayerStateManager: """ 玩家状态管理器。 - 支持并发读取（读锁）和写入（写锁）。 - 使用玩家级细粒度锁降低锁竞争。 - 提供增量更新（apply_patch）减少深拷贝。 """ def __init__(self, max_players=10000, idle_timeout=3600): self._global_lock = threading.RLock() # 保护内部结构（states 字典、玩家锁字典） self._player_locks: Dict[str, threading.RLock] = {} # 每个玩家独立的锁 self.states = OrderedDict() # name -> entry self.max_players = max_players self.idle_timeout = idle_timeout # 统计信息 self.access_count = 0 self.hit_count = 0 self.miss_count = 0 self._start_cleaner() # ---------- 玩家锁管理 ---------- def _get_player_lock(self, name: str) -> threading.RLock: """获取或创建玩家专用锁（线程安全）""" with self._global_lock: if name not in self._player_locks: self._player_locks[name] = threading.RLock() return self._player_locks[name] def _cleanup_player_lock(self, name: str): """清理无用的玩家锁（当玩家被移除时调用）""" with self._global_lock: self._player_locks.pop(name, None) # ---------- 核心接口 ---------- def get(self, name: str) -> Optional[Dict]: """ 获取玩家状态（返回引用，只读）。 调用方不应直接修改返回的状态，如需修改请使用 update 或 set。 """ # 使用玩家级锁（读操作） with self._get_player_lock(name): entry = self.states.get(name) if entry: entry['last_access'] = time.time() self.access_count += 1 self.hit_count += 1 return entry['state'] self.miss_count += 1 return None def get_ref(self, name: str) -> Optional[Dict]: """同 get，返回引用（只读）""" return self.get(name) def set(self, name: str, state: Dict, mark_dirty_flag=True): """ 直接设置状态（会深拷贝存储）。 若状态未发生实质变化，可跳过标记脏。 """ with self._get_player_lock(name): # 判断是否需要实际更新 old_entry = self.states.get(name) if old_entry and old_entry['state'] is state: # 同一个对象，无需操作 return # LRU 淘汰 if len(self.states) >= self.max_players and name not in self.states: self._evict_lru() new_state = self._deep_copy_state(state) new_version = (old_entry.get('version', 0) + 1) if old_entry else 0 self.states[name] = { 'state': new_state, 'last_access': time.time(), 'version': new_version, 'dirty': mark_dirty_flag } if mark_dirty_flag: mark_dirty(name) # 检测影响属性的变更，使属性缓存失效 if old_entry and self._stats_affected(old_entry['state'], new_state): from stats_cache import stats_version_manager stats_version_manager.invalidate(name) def update(self, name: str, update_func: Callable[[Dict], Optional[Dict]]) -> Optional[Dict]: """ 原子更新：持有玩家锁，执行 update_func，如果返回非 None 则更新状态。 update_func 接收当前状态的一个深拷贝副本，可安全修改。 返回新状态（引用），若更新失败返回 None。 """ with self._get_player_lock(name): entry = self.states.get(name) if not entry: # 尝试从数据库加载（注意：此处避免死锁，不能调用需要锁的函数） from db_storage import load_state_from_tables state = load_state_from_tables(name) if state is None: return None entry = { 'state': self._deep_copy_state(state), 'last_access': time.time(), 'version': 0, 'dirty': False } self.states[name] = entry old_state = entry['state'] # 传入深拷贝，保证 update_func 不会意外修改缓存 new_state = update_func(self._deep_copy_state(old_state)) if new_state is None: return None # 放弃更新 # 检查是否有实际变化（简单比较对象引用） if new_state is old_state: return old_state # 版本增加 & 脏标记 entry['state'] = self._deep_copy_state(new_state) entry['last_access'] = time.time() entry['version'] += 1 entry['dirty'] = True mark_dirty(name) # 检测属性变化 if self._stats_affected(old_state, new_state): from stats_cache import stats_version_manager stats_version_manager.invalidate(name) return entry['state'] def apply_patch(self, name: str, patch: Dict) -> Optional[Dict]: """ 增量更新：直接应用补丁到当前状态，避免全量拷贝。 patch 应包含需要修改的字段路径（支持嵌套）。 返回应用补丁后的状态，若玩家不存在则返回 None。 """ def patcher(state): self._apply_patch_recursive(state, patch) return state # 复用 update 框架，但注意 patcher 会直接修改传入的 state（浅层修改） # 为保证线程安全，update 内部会拷贝 state 再调用 patcher，因此无问题 return self.update(name, patcher) def _apply_patch_recursive(self, target: Dict, patch: Dict): """递归应用补丁""" for key, value in patch.items(): if isinstance(target.get(key), dict) and isinstance(value, dict): self._apply_patch_recursive(target[key], value) else: target[key] = value def update_batch(self, updates: Dict[str, Callable[[Dict], Optional[Dict]]]) -> Dict[str, Optional[Dict]]: """ 批量原子更新：依次持有各玩家锁，但注意避免死锁（按名称排序）。 返回字典 {player_name: new_state_or_None}。 """ # 按名称排序，避免死锁 sorted_names = sorted(updates.keys()) results = {} for name in sorted_names: update_func = updates[name] # 单独加锁更新（不会与其他玩家锁交叉） res = self.update(name, update_func) results[name] = res return results def get_batch(self, names: List[str]) -> Dict[str, Optional[Dict]]: """ 批量获取状态（引用），每个玩家单独加锁，不影响并发性。 """ result = {} for name in names: # 单独加锁获取 with self._get_player_lock(name): entry = self.states.get(name) if entry: entry['last_access'] = time.time() result[name] = entry['state'] else: result[name] = None return result def mark_clean(self, name: str): with self._get_player_lock(name): entry = self.states.get(name) if entry: entry['dirty'] = False clear_dirty(name) def get_dirty_states(self) -> Dict[str, Dict]: dirty_names = get_dirty_list() result = {} for name in dirty_names: with self._get_player_lock(name): entry = self.states.get(name) if entry and entry.get('dirty'): result[name] = entry['state'] return result def remove(self, name: str): with self._get_player_lock(name): if name in self.states: clear_dirty(name) from stats_cache import stats_version_manager stats_version_manager.remove(name) del self.states[name] self._cleanup_player_lock(name) # ---------- LRU 与空闲清理 ---------- def _evict_lru(self): """淘汰最久未访问的玩家（持有全局锁时调用）""" if not self.states: return sorted_items = sorted(self.states.items(), key=lambda x: x[1]['last_access']) evict_count = max(1, int(len(sorted_items) * 0.1)) from stats_cache import stats_version_manager for name, _ in sorted_items[:evict_count]: if name in self.states: del self.states[name] stats_version_manager.remove(name) clear_dirty(name) self._cleanup_player_lock(name) logger.debug(f"状态管理器LRU清理: 移除 {evict_count} 个玩家") def _start_cleaner(self): def cleaner(): while True: time.sleep(300) try: self._clean_idle_players() except Exception as e: logger.error(f"清理空闲玩家失败: {e}") thread = threading.Thread(target=cleaner, daemon=True) thread.start() def _clean_idle_players(self): now = time.time() to_remove = [] # 需遍历所有玩家，为避免长时间持锁，先获取 name 列表 with self._global_lock: names = list(self.states.keys()) for name in names: with self._get_player_lock(name): entry = self.states.get(name) if entry and now - entry['last_access'] > self.idle_timeout: to_remove.append(name) for name in to_remove: self.remove(name) if to_remove: logger.info(f"清理空闲玩家: {len(to_remove)} 个") # ---------- 统计与辅助 ---------- def get_stats(self) -> Dict: with self._global_lock: hit_rate = self.hit_count / max(1, self.access_count) * 100 return { 'players_in_memory': len(self.states), 'access_count': self.access_count, 'hit_count': self.hit_count, 'miss_count': self.miss_count, 'hit_rate': f"{hit_rate:.2f}%", 'max_players': self.max_players } @staticmethod def _stats_affected(old_state: Dict, new_state: Dict) -> bool: """检查哪些变更会影响最终属性""" if not old_state: return True affected_keys = [ ('player', 'level'), ('player', 'rebirth'), ('player', 'attackEnhance'), ('player', 'defenseEnhance'), ('player', 'expMultiplier'), ('player', 'gameSpeed'), ('player', 'accuracy'), ('player', 'dodge'), ('hero', 'level'), ('hero', 'rebirth'), ('hero', 'attackEnhance'), ('hero', 'defenseEnhance'), 'equipment', 'zodiac_equipped', 'temporary_buffs' ] for key in affected_keys: if isinstance(key, tuple): old_val = old_state.get(key[0], {}).get(key[1]) new_val = new_state.get(key[0], {}).get(key[1]) if old_val != new_val: return True else: if old_state.get(key) != new_state.get(key): return True return False @staticmethod def _deep_copy_state(state: Dict) -> Dict: """深拷贝状态（仅用于存储，避免外部修改污染缓存）""" if state is None: return None import copy return copy.deepcopy(state) player_state_manager = PlayerStateManager( max_players=50000, # 足够容纳所有活跃玩家 idle_timeout=7200 # 2小时空闲才清理 )