# Redis 使用规范

> **版本**: v1.0 | **修订日期**: 2026-07-15
> **技术栈**: Redis 7.0+ / redis-py (async)

---

## 一、连接管理

```python
# app/core/redis.py
import redis.asyncio as redis
from app.core.config import settings

redis_client = redis.from_url(
    settings.REDIS_URL,
    encoding="utf-8",
    decode_responses=True,
    max_connections=20,
)

async def get_redis() -> redis.Redis:
    return redis_client

# 生命周期管理
async def close_redis():
    await redis_client.close()
```

---

## 二、Key 命名规范

### 2.1 格式

```
{app_prefix}:{module}:{entity}:{identifier}
```

冒号 `:` 分隔层级，便于 Redis CLI 和可视化工具浏览。

### 2.2 示例

| 用途         | Key 示例                                    |
|--------------|---------------------------------------------|
| 用户缓存     | app:user:profile:123                        |
| 仓库列表     | app:warehouse:list:active                   |
| 会话         | app:session:{token}                         |
| 限流         | app:ratelimit:api:{user_id}:{endpoint}      |
| 分布式锁     | app:lock:inventory_sync:{warehouse_id}      |
| Celery 任务  | celery-task-meta-{task_id}（Celery 自动管理）|

### 2.3 规则

| 规则         | 说明                                           |
|--------------|------------------------------------------------|
| 前缀         | 所有 key 必须以 `app:` 开头                    |
| 层级         | 冒号分隔，不超过 4 层                          |
| 长度         | 不超过 128 字节                                |
| 禁止空格     | 使用下划线或驼峰代替空格                       |
| TTL          | 所有 key 必须设置过期时间                      |

---

## 三、缓存使用

### 3.1 装饰器模式

```python
import json
import functools

def cached(key_template: str, ttl: int = 300):
    """缓存装饰器"""
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            key = key_template.format(**kwargs)
            r = await get_redis()
            # 尝试读缓存
            data = await r.get(key)
            if data:
                return json.loads(data)
            # 执行函数
            result = await func(*args, **kwargs)
            # 写缓存
            await r.setex(key, ttl, json.dumps(result, default=str))
            return result
        return wrapper
    return decorator

# 使用
@cached(key_template="app:warehouse:detail:{warehouse_id}", ttl=600)
async def get_warehouse_detail(warehouse_id: int):
    return await warehouse_service.get_by_id(db, warehouse_id)
```

### 3.2 缓存失效策略

| 策略         | 适用场景           | 实现                               |
|--------------|--------------------|------------------------------------|
| TTL 过期     | 所有缓存           | setex 设置过期时间                 |
| 主动删除     | 数据更新时         | 更新/删除操作后 delete key         |
| 模式删除     | 批量失效           | scan + delete（禁止 keys *）       |
| 版本号       | 配置类数据         | key 中嵌入版本号，更新时递增       |

```python
# 缓存失效示例
async def invalidate_warehouse_cache(warehouse_id: int):
    r = await get_redis()
    await r.delete(f"app:warehouse:detail:{warehouse_id}")
    await r.delete("app:warehouse:list:active")
```

---

## 四、分布式锁

```python
from redis.asyncio.lock import Lock

async def with_inventory_lock(warehouse_id: int, func, *args, **kwargs):
    """带分布式锁的库存操作"""
    r = await get_redis()
    lock_key = f"app:lock:inventory:{warehouse_id}"
    lock = Lock(r, lock_key, timeout=30, blocking_timeout=5)

    async with lock:
        return await func(*args, **kwargs)

# 使用
await with_inventory_lock(warehouse_id=123, func=sync_inventory)
```

| 规则         | 说明                                           |
|--------------|------------------------------------------------|
| timeout      | 锁的最大持有时间，防止死锁                     |
| blocking     | 获取锁时是否阻塞等待                           |
| blocking_timeout | 阻塞等待的最大时间                         |
| 唯一标识     | Lock 自动使用 UUID，确保释放安全               |

---

## 五、限流

### 5.1 滑动窗口限流

```python
import time

async def is_rate_limited(user_id: int, endpoint: str, limit: int = 100, window: int = 60) -> bool:
    """滑动窗口限流"""
    r = await get_redis()
    key = f"app:ratelimit:{user_id}:{endpoint}"
    now = time.time()

    pipe = r.pipeline()
    pipe.zremrangebyscore(key, 0, now - window)
    pipe.zadd(key, {str(now): now})
    pipe.zcard(key)
    pipe.expire(key, window)
    results = await pipe.execute()

    return results[2] > limit
```

---

## 六、数据结构选择

| 场景           | 数据结构       | 示例                                  |
|----------------|----------------|---------------------------------------|
| 简单缓存       | String         | GET/SET user profile JSON             |
| 对象属性       | Hash           | HSET user:123 name "Alice"            |
| 列表/队列      | List           | LPUSH/RPOP 消息队列                   |
| 去重集合       | Set            | SADD user:123:tags "vip"              |
| 排行榜/排序    | Sorted Set     | ZADD leaderboard score member          |
| 位操作         | Bitmap         | SETBIT user:active:20260715 user_id 1 |
| 计数器         | String (INCR)  | INCR page:views:home                  |

---

## 七、序列化规范

| 规则         | 说明                                           |
|--------------|------------------------------------------------|
| 格式         | 统一使用 JSON                                  |
| 禁止 pickle  | 安全风险，禁止序列化 Python 对象               |
| datetime     | 序列化为 ISO 8601 字符串                       |
| Decimal      | 序列化为字符串                                 |
| bytes        | Base64 编码后存储                              |

---

## 八、内存管理

| 配置           | 推荐值          | 说明                               |
|----------------|-----------------|------------------------------------|
| maxmemory      | 根据服务器内存  | 设置最大内存限制                   |
| maxmemory-policy | allkeys-lru   | 内存满时淘汰策略                   |
| 过期时间       | 必须设置        | 所有 key 必须有 TTL                |
| 大 key 监控    | redis-cli --bigkeys | 定期检查大 key                 |

---

## 九、测试规范

```python
# tests/conftest.py
import fakeredis.aioredis

@pytest.fixture
async def redis_client():
    """使用 fakeredis 隔离测试"""
    fake = fakeredis.aioredis.FakeRedis(decode_responses=True)
    yield fake
    await fake.flushall()
    await fake.close()

# 测试示例
async def test_cache_hit(redis_client):
    await redis_client.setex("app:test:key", 60, "value")
    result = await redis_client.get("app:test:key")
    assert result == "value"
```

| 规则         | 说明                                           |
|--------------|------------------------------------------------|
| fakeredis    | 单元测试使用 fakeredis，不依赖真实 Redis       |
| 隔离         | 每个测试前后 flushall                          |
| 集成测试     | 使用独立 Redis DB（db=15）                     |