## 前言

缓存是提升系统性能最有效的手段。中台化后，多个业务线共享 Redis 集群，如何隔离、如何管理、如何设计多级缓存是关键。本篇讲解缓存中台的完整建设方案。

---

## 一、缓存中台全景

```
┌──────────────────────────────────────────────────────────────┐
│                      业务中台 / 前台                         │
│  用户中心    订单中心    支付中心    商品中心               │
└──────────┬──────────┬──────────┬──────────┬─────────────────┘
           │          │          │          │
┌──────────┴──────────┴──────────┴──────────┴─────────────────┐
│                    缓存中台                                  │
│  ┌──────────────────────────────────────────────────────┐   │
│  │             本地缓存（L1）                           │   │
│  │  Caffeine / Guava Cache / Go Cache                  │   │
│  └────────────────────┬─────────────────────────────────┘   │
│                       │                                    │
│  ┌────────────────────┴─────────────────────────────────┐   │
│  │             分布式缓存（L2）                         │   │
│  │  Redis Cluster (6+ 节点)                             │   │
│  │  ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ │ │
│  │  │Shard0│ │Shard1│ │Shard2│ │Shard0│ │Shard1│ │Shard2│ │ │
│  │  │Master│ │Master│ │Master│ │Slave │ │Slave │ │Slave │ │ │
│  │  └──────┘ └──────┘ └──────┘ └──────┘ └──────┘ └──────┘ │ │
│  └──────────────────────────────────────────────────────┘   │
└──────────────────────────────────────────────────────────────┘
```

---

## 二、Redis Cluster 部署

### 集群架构

```
Redis Cluster（6 节点，3 主 3 从）：

  Slot 分配（16384 个槽位）：
  ┌──────────┐ ┌──────────┐ ┌──────────┐
  │ Master 0 │ │ Master 1 │ │ Master 2 │
  │ Slots    │ │ Slots    │ │ Slots    │
  │ 0-5460   │ │ 5461-10922│ │10923-16383│
  └────┬─────┘ └────┬─────┘ └────┬─────┘
       │            │            │
  ┌────┴─────┐ ┌────┴─────┐ ┌────┴─────┐
  │ Slave 0  │ │ Slave 1  │ │ Slave 2  │
  │ (复制M0) │ │ (复制M1) │ │ (复制M2) │
  └──────────┘ └──────────┘ └──────────┘

  故障转移：
  - Master 0 挂 → Slave 0 升级为 Master
  - 自动 failover
```

### Docker 部署

```yaml
# docker-compose.yml
version: '3'

services:
  redis-node-1:
    image: redis:7.2
    command: redis-server --port 7000 --cluster-enabled yes --cluster-config-file nodes.conf --cluster-node-timeout 5000 --appendonly yes --cluster-announce-ip 127.0.0.1 --cluster-announce-port 7000
    ports:
      - "7000:7000"
  
  redis-node-2:
    image: redis:7.2
    command: redis-server --port 7001 --cluster-enabled yes --cluster-config-file nodes.conf --cluster-node-timeout 5000 --appendonly yes --cluster-announce-ip 127.0.0.1 --cluster-announce-port 7001
    ports:
      - "7001:7001"
  
  # ... 4 more nodes
  
  redis-cluster-init:
    image: redis:7.2
    command: redis-cli --cluster create 127.0.0.1:7000 127.0.0.1:7001 127.0.0.1:7002 127.0.0.1:7003 127.0.0.1:7004 127.0.0.1:7005 --cluster-replicas 1 --cluster-yes
    depends_on:
      - redis-node-1
      - redis-node-2
```

### K8s 部署

```yaml
# StatefulSet
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: redis
  namespace: cache
spec:
  serviceName: redis
  replicas: 6
  podManagementPolicy: Parallel
  template:
    spec:
      containers:
        - name: redis
          image: redis:7.2
          command:
            - redis-server
            - --port $(REDIS_PORT)
            - --cluster-enabled yes
            - --cluster-config-file nodes.conf
            - --cluster-node-timeout 5000
            - --appendonly yes
          env:
            - name: REDIS_PORT
              value: "6379"
          resources:
            requests:
              cpu: 500m
              memory: 1Gi
            limits:
              cpu: 1
              memory: 2Gi
          volumeMounts:
            - name: data
              mountPath: /data
  volumeClaimTemplates:
    - metadata:
        name: data
      spec:
        accessModes: ["ReadWriteOnce"]
        resources:
          requests:
            storage: 50Gi
```

---

## 三、多租户隔离

### 方案 1：Key 前缀

```python
# 最简单的多租户隔离：Key 前缀
class CacheClient:
    def __init__(self, team):
        self.prefix = f"team:{team}:"
        self.redis = RedisCluster(host='redis-cluster', port=7000)
    
    def get(self, key):
        return self.redis.get(self.prefix + key)
    
    def set(self, key, value, ttl=3600):
        return self.redis.set(self.prefix + key, value, ex=ttl)

# 使用
payments_cache = CacheClient("payments")
orders_cache = CacheClient("orders")

payments_cache.set("user:123", data)  # → team:payments:user:123
orders_cache.set("user:123", data)    # → team:orders:user:123
```

### 方案 2：独立 DB

```python
# Redis Cluster 不支持多 DB，但独立实例可以
# 每个 Team 一个 Redis 实例
team_configs = {
    "payments": {"host": "redis-payments", "db": 0},
    "orders": {"host": "redis-orders", "db": 0},
    "users": {"host": "redis-users", "db": 0},
}
```

### 方案 3：Proxy 隔离

```yaml
# 用 Twemproxy 或 Redis Proxy 做多租户
# 统一入口，按 Key 前缀路由到不同后端
proxy_config:
  listen: 0.0.0.0:6379
  pools:
    payments:
      servers:
        - redis-payments-1:6379:1
        - redis-payments-2:6379:1
      key_pattern: "team:payments:*"
    
    orders:
      servers:
        - redis-orders-1:6379:1
        - redis-orders-2:6379:1
      key_pattern: "team:orders:*"
```

---

## 四、多级缓存架构

### L1 + L2 架构

```python
# L1: 本地缓存（Caffeine/Guava）
# L2: 分布式缓存（Redis）
# L3: 数据库

class MultiLevelCache:
    def __init__(self):
        # L1: 本地缓存
        self.l1 = Caffeine(
            maximum_size=10000,
            expire_after_write=300,  # 5 分钟
        )
        # L2: Redis
        self.l2 = RedisCluster(host='redis-cluster', port=7000)
        # 消息通知
        self.mq = MQClient()
    
    def get(self, key):
        # L1 查询
        value = self.l1.get(key)
        if value:
            return value  # 命中 L1
        
        # L2 查询
        value = self.l2.get(key)
        if value:
            self.l1.set(key, value)  # 回填 L1
            return value  # 命中 L2
        
        # 数据库查询
        value = db.query(key)
        if value:
            self.l1.set(key, value)
            self.l2.set(key, value, ttl=3600)
            return value
        
        return None  # 未命中
    
    def set(self, key, value, ttl=3600):
        self.l1.set(key, value)
        self.l2.set(key, value, ttl=ttl)
        # 通知其他实例更新 L1
        self.mq.publish('cache-update', {'key': key})
    
    def invalidate(self, key):
        self.l1.delete(key)
        self.l2.delete(key)
        # 通知其他实例删除 L1
        self.mq.publish('cache-invalidate', {'key': key})
    
    def on_cache_event(self, event):
        """监听缓存失效事件"""
        if event['type'] == 'cache-update':
            self.l1.delete(event['key'])  # 删除 L1，下次从 L2 加载
        elif event['type'] == 'cache-invalidate':
            self.l1.delete(event['key'])
            self.l2.delete(event['key'])
```

### 缓存一致性

```python
# Cache Aside 模式（推荐）
def update_data(key, new_value):
    # 1. 更新数据库
    db.update(key, new_value)
    # 2. 删除缓存
    cache.invalidate(key)

def get_data(key):
    # 1. 查缓存
    value = cache.get(key)
    if value:
        return value
    # 2. 查数据库
    value = db.query(key)
    if value:
        cache.set(key, value, ttl=300)
    return value

# 为什么删除缓存而不是更新缓存？
# 1. 避免并发更新导致数据不一致
# 2. 有些缓存值需要复杂计算，直接删除更高效
# 3. 延迟双删策略可以进一步保证一致性
```

### 延迟双删

```python
import threading
import time

def update_with_double_delete(key, new_value):
    # 1. 先删缓存
    cache.delete(key)
    # 2. 更新数据库
    db.update(key, new_value)
    # 3. 延迟再删一次
    threading.Timer(0.5, cache.delete, args=[key]).start()
```

---

## 五、缓存设计模式

### 1. 缓存穿透

```python
# 问题：查询不存在的 Key → 每次都查数据库
# 解决：缓存空值

def get_user(user_id):
    cache_key = f"user:{user_id}"
    value = cache.get(cache_key)
    
    if value is not None:
        if value == "NULL":
            return None  # 缓存的空值
        return value
    
    # 查数据库
    user = db.get_user(user_id)
    if user:
        cache.set(cache_key, user, ttl=300)
    else:
        # 缓存空值，短 TTL
        cache.set(cache_key, "NULL", ttl=60)
    
    return user

# 更进一步：布隆过滤器
from pybloom_live import ScalableBloomFilter

bloom = ScalableBloomFilter(initial_capacity=1000000, error_rate=0.001)

def init_bloom():
    all_user_ids = db.get_all_user_ids()
    for uid in all_user_ids:
        bloom.add(uid)

def get_user_safe(user_id):
    if user_id not in bloom:
        return None  # 布隆过滤器说不存在
    return get_user(user_id)
```

### 2. 缓存击穿

```python
# 问题：热点 Key 过期 → 大量请求打到数据库
# 解决：互斥锁 + 永不过期

# 方案 1：互斥锁
def get_hot_data(key):
    value = cache.get(key)
    if value:
        return value
    
    # 获取锁
    lock_key = f"lock:{key}"
    if cache.set(lock_key, "1", nx=True, ex=10):
        try:
            value = db.query(key)
            cache.set(key, value, ttl=3600)
            return value
        finally:
            cache.delete(lock_key)
    else:
        # 等待并重试
        time.sleep(0.1)
        return get_hot_data(key)

# 方案 2：逻辑过期
def set_with_logical_expiry(key, value, ttl):
    data = {
        'value': value,
        'expire_at': time.time() + ttl,
    }
    cache.set(key, json.dumps(data))  # 不设 TTL

def get_with_logical_expiry(key):
    data = cache.get(key)
    if not data:
        return None
    data = json.loads(data)
    if time.time() < data['expire_at']:
        return data['value']  # 未过期
    # 已过期，异步更新
    if cache.set(f"lock:{key}", "1", nx=True, ex=10):
        threading.Thread(target=refresh_cache, args=(key,)).start()
    return data['value']  # 返回旧值
```

### 3. 缓存雪崩

```python
# 问题：大量 Key 同时过期 → 集中打数据库
# 解决：随机 TTL + 多级缓存

def set_with_jitter(key, value, base_ttl=3600):
    jitter = random.randint(0, 300)  # 0-5 分钟随机
    cache.set(key, value, ttl=base_ttl + jitter)
```

---

## 六、Redis 最佳实践

### 数据结构选择

```python
# String：简单 KV
cache.set("user:123", json.dumps(user_data))

# Hash：对象属性
cache.hset("user:123", "name", "张三")
cache.hset("user:123", "age", "30")
# 比 String 节省内存，支持部分更新

# List：队列/排行榜
cache.lpush("queue:orders", order_id)
cache.zadd("rank:daily", {user_id: score})

# Set：去重/集合运算
cache.sadd("active:users:20240115", user_id)
cache.sinter("active:users:20240115", "vip:users")  # 交集

# Sorted Set：排行榜
cache.zadd("leaderboard", {player_id: score})
cache.zrevrange("leaderboard", 0, 9)  # Top 10

# HyperLogLog：基数统计
cache.pfadd("unique:visitors:20240115", user_id)
cache.pfcount("unique:visitors:20240115")  # 去重 UV

# Bitmap：状态标记
cache.setbit("login:20240115", user_id, 1)
cache.bitcount("login:20240115")  # 当天登录人数
```

### Pipeline 批量操作

```python
# 逐条操作（慢）
for key in keys:
    cache.get(key)  # N 次网络往返

# Pipeline 批量（快）
pipe = cache.pipeline()
for key in keys:
    pipe.get(key)
results = pipe.execute()  # 1 次网络往返
```

### Lua 脚本

```python
# 原子操作用 Lua
lua_script = """
local key = KEYS[1]
local value = ARGV[1]
local ttl = ARGV[2]

local current = redis.call('GET', key)
if not current then
    redis.call('SET', key, value, 'EX', ttl)
    return 1
else
    return 0
end
"""

# 执行原子操作
cache.eval(lua_script, 1, "lock:123", "owner", 10)
```

### 内存优化

```yaml
# Redis 配置优化
maxmemory: 4gb
maxmemory-policy: allkeys-lru  # LRU 淘汰
# 或 volatile-lru（只淘汰有过期时间的）
# 或 allkeys-lfu（LFU 淘汰）

# 压缩
# Redis 7.0+ 支持列表打包
list-max-listpack-size: -2  # 8KB 以下用 listpack
```

---

## 七、监控告警

```yaml
# Redis Exporter
scrape_configs:
  - job_name: 'redis'
    static_configs:
      - targets: ['redis-exporter:9121']

# 关键指标
# redis_memory_used_bytes → 内存使用
# redis_connected_clients → 连接数
# redis_commands_processed_total → QPS
# redis_keyspace_hits_total / redis_keyspace_misses_total → 命中率
```

```yaml
groups:
  - name: redis-alerts
    rules:
      - alert: RedisMemoryHigh
        expr: redis_memory_used_bytes / redis_memory_max_bytes * 100 > 90
        for: 5m
        labels:
          severity: warning
      
      - alert: RedisHighConnections
        expr: redis_connected_clients > 1000
        for: 5m
        labels:
          severity: warning
      
      - alert: RedisLowHitRate
        expr: |
          redis_keyspace_hits_total / (redis_keyspace_hits_total + redis_keyspace_misses_total) * 100 < 80
        for: 10m
        labels:
          severity: info
      
      - alert: RedisClusterDown
        expr: redis_cluster_state != 1
        for: 1m
        labels:
          severity: critical
```

> ⚠️ **踩坑提示**：
> - Redis Cluster 不支持跨 Slot 的事务
> - 大 Key（> 10KB）影响性能，需拆分
> - 热点 Key 需要本地缓存兜底
> - 缓存更新用删除而非更新
> - 永远设置 TTL，避免内存泄漏

---

## 要点回顾

| 维度 | 要点 |
|------|------|
| 集群 | Redis Cluster 3 主 3 从 |
| 多租户 | Key 前缀 / 独立实例 / Proxy |
| 多级缓存 | L1 本地 + L2 Redis |
| 一致性 | Cache Aside + 延迟双删 |
| 穿透 | 缓存空值 + 布隆过滤器 |
| 击穿 | 互斥锁 + 逻辑过期 |
| 雪崩 | 随机 TTL + 多级缓存 |

- Redis Cluster 6 节点起步，3 主 3 从
- 多租户用 Key 前缀最简单
- 多级缓存 L1 + L2 兼顾性能和一致性
- Cache Aside 模式：更新 DB → 删除缓存
- 用 Pipeline 和 Lua 提升性能

---

## 下一篇预告

下一篇 **【中台·技术篇】配置中心中台：Apollo 与 Nacos 选型与多环境配置管理** 将讲解配置中心的中台化。

缓存是提升系统性能最有效的手段。中台化后，多个业务线共享 Redis 集群，如何隔离、如何管理、如何设计多级缓存是关键。本篇讲解缓存中台的完整建设方案。

【中台·技术篇】缓存中台建设：Redis 集群统一管理与多级缓存架构

研发部经理

人工智能

缓存中台建设方案详解：涵盖Redis Cluster集群部署、多租户隔离策略、L1本地+L2分布式多级缓存架构、缓存穿透/击穿/雪崩解决方案，以及Pipeline优化、监控告警与最佳实践，助力业务线高效共享缓存资源。

2026上云采购 | AI焕新·智启新局

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【中台·技术篇】缓存中台建设：Redis 集群统一管理与多级缓存架构-腾讯云开发者社区-腾讯云

【中台·技术篇】缓存中台建设：Redis 集群统一管理与多级缓存架构

【中台·技术篇】缓存中台建设：Redis 集群统一管理与多级缓存架构

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐