import time import redis from typing import List, Dict, Callable, Any class SafeRedisClient: def __init__(self, redis_conf: Dict, logger: Callable[[str], None] = None): safe_conf = dict(redis_conf) safe_conf.setdefault('socket_timeout', 5.0) # 读写超时防死等 safe_conf.setdefault('socket_connect_timeout', 5.0) # 建连超时防死等 safe_conf.setdefault('socket_keepalive', True) # TCP 操作系统级保活 safe_conf.setdefault('health_check_interval', 30) # 自动探测断连并重连 self._client = redis.Redis(**safe_conf) self._logger = logger def _log(self, msg: str): if self._logger: self._logger(f"[Safe-Redis] {msg}") else: print(f"[Safe-Redis] {msg}") def _safe_execute(self, func, default_return, *args, **kwargs): """通用安全执行器:遇错自动重试一次,彻底失败返回安全默认值""" try: return func(*args, **kwargs) except Exception as e: func_name = func.__name__ if hasattr(func, '__name__') else 'operation' self._log(f"Attempt 1 failed ({func_name}): {e}. Retrying in 0.5s...") time.sleep(0.5) try: return func(*args, **kwargs) except Exception as e2: self._log(f"Attempt 2 failed ({func_name}): {e2}. Returning safe fallback: {default_return}.") return default_return def exists(self, name) -> bool: return bool(self._safe_execute(self._client.exists, False, name)) def get(self, name) -> Any: return self._safe_execute(self._client.get, None, name) def set(self, name, value, ex=None) -> bool: return self._safe_execute(self._client.set, False, name, value, ex=ex) def setex(self, name, time_s, value) -> bool: return self._safe_execute(self._client.setex, False, name, time_s, value) def delete(self, *names) -> int: if not names: return 0 return self._safe_execute(self._client.delete, 0, *names) def incr(self, name, amount=1) -> int: # 如果增加失败,默认返回 1,防止触发外部数学运算崩溃(同时保证退避策略能生效) return self._safe_execute(self._client.incr, 1, name, amount=amount) def zadd(self, name, mapping) -> int: if not mapping: return 0 return self._safe_execute(self._client.zadd, 0, name, mapping) def zrem(self, name, *values) -> int: if not values: return 0 return self._safe_execute(self._client.zrem, 0, name, *values) def bulk_zadd(self, name, mapping) -> bool: """封装 Pipeline,用于一次性安全地写入大量 zset 元素""" if not mapping: return True def _op(): pipe = self._client.pipeline() for k, v in mapping.items(): pipe.zadd(name, {k: v}) pipe.execute() return True return self._safe_execute(_op, False)