import redis import uuid from typing import Optional from app.config import settings from app.logger import get_redis_logger logger = get_redis_logger() class RedisLock: """Redis分布式锁""" def __init__(self, redis_client: redis.Redis, key: str, timeout: int = None): self.redis_client = redis_client self.key = f"lock:{key}" self.timeout = timeout or settings.redis_lock_timeout self.identifier = str(uuid.uuid4()) self.acquired = False def acquire(self, blocking: bool = True, timeout: int = None) -> bool: """获取分布式锁""" logger.debug(f"🔒 尝试获取Redis锁: {self.key}") lua_script = """ if redis.call("GET", KEYS[1]) == false then return redis.call("SETEX", KEYS[1], ARGV[1], ARGV[2]) else return false end """ result = self.redis_client.eval( lua_script, 1, self.key, self.timeout, self.identifier ) self.acquired = bool(result) if self.acquired: logger.debug(f"✅ Redis锁获取成功: {self.key}") else: logger.debug(f"❌ Redis锁获取失败: {self.key}") return self.acquired def release(self) -> bool: """释放分布式锁""" if not self.acquired: return False lua_script = """ if redis.call("GET", KEYS[1]) == ARGV[1] then return redis.call("DEL", KEYS[1]) else return 0 end """ result = self.redis_client.eval( lua_script, 1, self.key, self.identifier ) self.acquired = False return bool(result) def __enter__(self): """同步上下文管理器入口""" retries = 0 while retries < settings.redis_lock_max_retries: if self.acquire(blocking=False): return self import time time.sleep(settings.redis_lock_retry_delay) retries += 1 raise TimeoutError(f"Failed to acquire lock {self.key} after {retries} retries") def __exit__(self, exc_type, exc_val, exc_tb): """同步上下文管理器出口""" self.release() class RedisManager: def __init__(self): self.redis_client: Optional[redis.Redis] = None def connect(self) -> bool: """连接Redis Returns: bool: 连接是否成功 """ try: logger.info(f"🔴 正在连接Redis: {settings.redis_url}") self.redis_client = redis.from_url( settings.redis_url, encoding="utf-8", decode_responses=True, socket_connect_timeout=5, socket_timeout=5, retry_on_timeout=True ) result = self.redis_client.ping() if result: logger.info("✅ Redis连接成功") return True else: logger.error("❌ Redis ping失败") return False except Exception as e: logger.error(f"❌ Redis连接失败: {e}") return False def disconnect(self): """断开Redis连接""" if self.redis_client: self.redis_client.close() def create_lock(self, key: str, timeout: int = None) -> RedisLock: """创建分布式锁""" if not self.redis_client: raise RuntimeError("Redis client not connected") return RedisLock(self.redis_client, key, timeout) # 全局Redis管理器实例 redis_manager = RedisManager()