|
|
@@ -4,9 +4,9 @@ Redis缓存管理服务
|
|
|
"""
|
|
|
import json
|
|
|
import time
|
|
|
-import asyncio
|
|
|
-from typing import Dict, List, Optional
|
|
|
+from typing import Awaitable, Callable, Dict, List, Optional
|
|
|
import logging
|
|
|
+import redis.asyncio as aioredis
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@@ -14,14 +14,16 @@ logger = logging.getLogger(__name__)
|
|
|
class RedisCacheService:
|
|
|
"""Redis缓存服务"""
|
|
|
|
|
|
- def __init__(self, redis_client):
|
|
|
+ def __init__(self, redis_provider: Callable[[], Awaitable[aioredis.Redis]]):
|
|
|
"""
|
|
|
初始化Redis缓存服务
|
|
|
|
|
|
Args:
|
|
|
- redis_client: redis.asyncio.Redis 实例
|
|
|
+ redis_provider: 一个可调用对象,返回 redis.asyncio.Redis 实例
|
|
|
"""
|
|
|
- self.redis = redis_client
|
|
|
+ if not callable(redis_provider):
|
|
|
+ raise ValueError("redis_provider 必须可调用")
|
|
|
+ self._redis_provider = redis_provider
|
|
|
|
|
|
# 缓存Key设计
|
|
|
self.KEY_USER_CONTEXT = "context:{user_id}" # 用户上下文
|
|
|
@@ -34,6 +36,13 @@ class RedisCacheService:
|
|
|
self.TTL_MESSAGES = 86400 # 24小时
|
|
|
self.TTL_SUMMARY = 604800 # 7天
|
|
|
|
|
|
+ async def _get_redis(self) -> aioredis.Redis:
|
|
|
+ """获取最新的Redis客户端,确保连接有效"""
|
|
|
+ redis_client = await self._redis_provider()
|
|
|
+ if not redis_client:
|
|
|
+ raise RuntimeError("Redis 客户端不可用")
|
|
|
+ return redis_client
|
|
|
+
|
|
|
async def get_context(self, user_id: str) -> Optional[Dict]:
|
|
|
"""
|
|
|
从Redis获取用户上下文(零延迟)
|
|
|
@@ -47,8 +56,9 @@ class RedisCacheService:
|
|
|
}
|
|
|
"""
|
|
|
try:
|
|
|
+ redis = await self._get_redis()
|
|
|
key = self.KEY_USER_CONTEXT.format(user_id=user_id)
|
|
|
- cached = await self.redis.get(key)
|
|
|
+ cached = await redis.get(key)
|
|
|
|
|
|
if cached:
|
|
|
context = json.loads(cached)
|
|
|
@@ -77,10 +87,11 @@ class RedisCacheService:
|
|
|
ttl: 过期时间(秒),默认24小时
|
|
|
"""
|
|
|
try:
|
|
|
+ redis = await self._get_redis()
|
|
|
key = self.KEY_USER_CONTEXT.format(user_id=user_id)
|
|
|
ttl = ttl or self.TTL_CONTEXT
|
|
|
|
|
|
- await self.redis.setex(
|
|
|
+ await redis.setex(
|
|
|
key,
|
|
|
ttl,
|
|
|
json.dumps(context, ensure_ascii=False)
|
|
|
@@ -106,6 +117,7 @@ class RedisCacheService:
|
|
|
使用Redis List结构,支持高效追加和范围查询
|
|
|
"""
|
|
|
try:
|
|
|
+ redis = await self._get_redis()
|
|
|
key = self.KEY_USER_MESSAGES.format(user_id=user_id)
|
|
|
|
|
|
message = {
|
|
|
@@ -116,13 +128,13 @@ class RedisCacheService:
|
|
|
}
|
|
|
|
|
|
# 追加到List尾部
|
|
|
- await self.redis.rpush(key, json.dumps(message, ensure_ascii=False))
|
|
|
+ await redis.rpush(key, json.dumps(message, ensure_ascii=False))
|
|
|
|
|
|
# 设置过期时间
|
|
|
- await self.redis.expire(key, self.TTL_MESSAGES)
|
|
|
+ await redis.expire(key, self.TTL_MESSAGES)
|
|
|
|
|
|
# 限制List长度(最多保留200条消息)
|
|
|
- await self.redis.ltrim(key, -200, -1)
|
|
|
+ await redis.ltrim(key, -200, -1)
|
|
|
|
|
|
logger.debug(f"✅ Redis追加消息: user={user_id}, round={round_number}, role={role}")
|
|
|
return True
|
|
|
@@ -144,10 +156,11 @@ class RedisCacheService:
|
|
|
count: 消息数量(默认30条,即15轮)
|
|
|
"""
|
|
|
try:
|
|
|
+ redis = await self._get_redis()
|
|
|
key = self.KEY_USER_MESSAGES.format(user_id=user_id)
|
|
|
|
|
|
# 获取最后N条消息
|
|
|
- messages_raw = await self.redis.lrange(key, -count, -1)
|
|
|
+ messages_raw = await redis.lrange(key, -count, -1)
|
|
|
|
|
|
messages = []
|
|
|
for msg_bytes in messages_raw:
|
|
|
@@ -169,8 +182,9 @@ class RedisCacheService:
|
|
|
async def get_current_round(self, user_id: str) -> int:
|
|
|
"""获取当前轮次"""
|
|
|
try:
|
|
|
+ redis = await self._get_redis()
|
|
|
key = self.KEY_USER_ROUND.format(user_id=user_id)
|
|
|
- round_bytes = await self.redis.get(key)
|
|
|
+ round_bytes = await redis.get(key)
|
|
|
|
|
|
if round_bytes:
|
|
|
round_str = round_bytes.decode('utf-8') if isinstance(round_bytes, bytes) else round_bytes
|
|
|
@@ -184,8 +198,9 @@ class RedisCacheService:
|
|
|
async def set_current_round(self, user_id: str, round_number: int) -> bool:
|
|
|
"""设置当前轮次"""
|
|
|
try:
|
|
|
+ redis = await self._get_redis()
|
|
|
key = self.KEY_USER_ROUND.format(user_id=user_id)
|
|
|
- await self.redis.setex(key, self.TTL_CONTEXT, str(round_number))
|
|
|
+ await redis.setex(key, self.TTL_CONTEXT, str(round_number))
|
|
|
return True
|
|
|
except Exception as e:
|
|
|
logger.error(f"❌ Redis设置轮次失败: {e}")
|
|
|
@@ -194,14 +209,15 @@ class RedisCacheService:
|
|
|
async def increment_round(self, user_id: str) -> int:
|
|
|
"""轮次自增"""
|
|
|
try:
|
|
|
+ redis = await self._get_redis()
|
|
|
key = self.KEY_USER_ROUND.format(user_id=user_id)
|
|
|
- new_round = await self.redis.incr(key)
|
|
|
- await self.redis.expire(key, self.TTL_CONTEXT)
|
|
|
+ new_round = await redis.incr(key)
|
|
|
+ await redis.expire(key, self.TTL_CONTEXT)
|
|
|
|
|
|
# 同时更新 user:{user_id}:meta 中的 current_round(保持一致性)
|
|
|
meta_key = f"user:{user_id}:meta"
|
|
|
- await self.redis.hset(meta_key, 'current_round', new_round)
|
|
|
- await self.redis.expire(meta_key, self.TTL_CONTEXT)
|
|
|
+ await redis.hset(meta_key, 'current_round', new_round)
|
|
|
+ await redis.expire(meta_key, self.TTL_CONTEXT)
|
|
|
|
|
|
logger.debug(f"✅ Redis轮次自增: user={user_id}, new_round={new_round}")
|
|
|
return new_round
|
|
|
@@ -213,8 +229,9 @@ class RedisCacheService:
|
|
|
async def get_summary(self, user_id: str, end_round: int) -> Optional[str]:
|
|
|
"""获取历史总结"""
|
|
|
try:
|
|
|
+ redis = await self._get_redis()
|
|
|
key = self.KEY_USER_SUMMARY.format(user_id=user_id, end_round=end_round)
|
|
|
- summary_bytes = await self.redis.get(key)
|
|
|
+ summary_bytes = await redis.get(key)
|
|
|
|
|
|
if summary_bytes:
|
|
|
summary = summary_bytes.decode('utf-8') if isinstance(summary_bytes, bytes) else summary_bytes
|
|
|
@@ -235,8 +252,9 @@ class RedisCacheService:
|
|
|
) -> bool:
|
|
|
"""保存历史总结"""
|
|
|
try:
|
|
|
+ redis = await self._get_redis()
|
|
|
key = self.KEY_USER_SUMMARY.format(user_id=user_id, end_round=end_round)
|
|
|
- await self.redis.setex(key, self.TTL_SUMMARY, summary)
|
|
|
+ await redis.setex(key, self.TTL_SUMMARY, summary)
|
|
|
|
|
|
logger.info(f"✅ Redis保存总结: user={user_id}, end_round={end_round}, 长度={len(summary)}字")
|
|
|
return True
|
|
|
@@ -394,8 +412,9 @@ class RedisCacheService:
|
|
|
async def invalidate_context(self, user_id: str) -> bool:
|
|
|
"""清除用户上下文缓存(强制刷新)"""
|
|
|
try:
|
|
|
+ redis = await self._get_redis()
|
|
|
key = self.KEY_USER_CONTEXT.format(user_id=user_id)
|
|
|
- await self.redis.delete(key)
|
|
|
+ await redis.delete(key)
|
|
|
logger.info(f"🗑️ Redis清除缓存: user={user_id}")
|
|
|
return True
|
|
|
except Exception as e:
|
|
|
@@ -412,13 +431,14 @@ class RedisCacheService:
|
|
|
current_round: 当前轮次
|
|
|
"""
|
|
|
try:
|
|
|
+ redis = await self._get_redis()
|
|
|
# 清空旧数据
|
|
|
messages_key = self.KEY_USER_MESSAGES.format(user_id=user_id)
|
|
|
- await self.redis.delete(messages_key)
|
|
|
+ await redis.delete(messages_key)
|
|
|
|
|
|
# 批量写入消息
|
|
|
if messages:
|
|
|
- pipeline = self.redis.pipeline()
|
|
|
+ pipeline = redis.pipeline()
|
|
|
for msg in messages:
|
|
|
pipeline.rpush(
|
|
|
messages_key,
|
|
|
@@ -440,7 +460,8 @@ class RedisCacheService:
|
|
|
async def health_check(self) -> bool:
|
|
|
"""健康检查"""
|
|
|
try:
|
|
|
- await self.redis.ping()
|
|
|
+ redis = await self._get_redis()
|
|
|
+ await redis.ping()
|
|
|
return True
|
|
|
except Exception as e:
|
|
|
logger.error(f"❌ Redis健康检查失败: {e}")
|