131 lines
3.7 KiB
Python
131 lines
3.7 KiB
Python
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() |