redis_client.py 2.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293
  1. """Redis 客户端 + Redis Key 命名空间。
  2. 提供进程级单例的 async Redis 连接池,以及统一的 key 前缀生成。
  3. """
  4. from __future__ import annotations
  5. import logging
  6. from typing import Optional
  7. from redis.asyncio import Redis, from_url
  8. from app.core.config import settings
  9. logger = logging.getLogger(__name__)
  10. _redis: Optional[Redis] = None
  11. def _k(suffix: str) -> str:
  12. """构造带前缀的 Redis key。"""
  13. return f"{settings.redis_key_prefix}:{suffix}"
  14. # 统一管理的 Redis Key
  15. KEY_CACHE_DATA = _k("cache:data") # JSON 缓存数据
  16. KEY_LOCK_LEADER = _k("lock:leader") # leader 选举锁(值=持有者标识)
  17. KEY_LOCK_REFRESH = _k("lock:refresh") # 防并发刷新锁
  18. async def get_redis() -> Redis:
  19. """获取进程级 Redis 单例(懒初始化)。"""
  20. global _redis
  21. if _redis is None:
  22. _redis = from_url(
  23. settings.redis_url,
  24. username=settings.redis_username,
  25. password=settings.redis_password,
  26. encoding="utf-8",
  27. decode_responses=True,
  28. )
  29. logger.info("Redis 连接已建立: %s", settings.redis_url)
  30. return _redis
  31. async def close_redis() -> None:
  32. """关闭 Redis 连接(应用关闭时调用)。"""
  33. global _redis
  34. if _redis is not None:
  35. try:
  36. await _redis.aclose()
  37. except Exception as exc: # noqa: BLE001
  38. logger.warning("关闭 Redis 连接异常: %s", exc)
  39. _redis = None
  40. async def try_acquire_lock(
  41. key: str, value: str, ttl_seconds: int
  42. ) -> bool:
  43. """尝试获取分布式锁(SET NX EX)。成功返回 True。"""
  44. r = await get_redis()
  45. ok = await r.set(key, value, nx=True, ex=ttl_seconds)
  46. return bool(ok)
  47. async def renew_lock(key: str, value: str, ttl_seconds: int) -> bool:
  48. """续期锁(仅当 value 与持有者一致时才续期,避免误续他人锁)。
  49. 使用 Lua 脚本保证原子性。
  50. """
  51. r = await get_redis()
  52. script = """
  53. if redis.call('get', KEYS[1]) == ARGV[1] then
  54. return redis.call('expire', KEYS[1], ARGV[2])
  55. else
  56. return 0
  57. end
  58. """
  59. result = await r.eval(script, 1, key, value, ttl_seconds)
  60. return bool(result)
  61. async def release_lock(key: str, value: str) -> bool:
  62. """释放锁(仅当持有者匹配时才删除)。"""
  63. r = await get_redis()
  64. script = """
  65. if redis.call('get', KEYS[1]) == ARGV[1] then
  66. return redis.call('del', KEYS[1])
  67. else
  68. return 0
  69. end
  70. """
  71. result = await r.eval(script, 1, key, value)
  72. return bool(result)