coordinate/utils/safe_redis_cli.py

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)