【系统架构设计百科】缓存架构：从本地到分布式的多级缓存体系

public class ProductService {

    private final RedisTemplate<String, Product> cache;
    private final ProductRepository db;

    public Product getProduct(Long id) {
        String key = "product:" + id;
        // 1. 先查缓存
        Product product = cache.opsForValue().get(key);
        if (product != null) {
            return product;
        }
        // 2. 缓存未命中，查数据库
        product = db.findById(id).orElse(null);
        if (product != null) {
            // 3. 回填缓存，设置过期时间
            cache.opsForValue().set(key, product, 30, TimeUnit.MINUTES);
        }
        return product;
    }

    public void updateProduct(Long id, Product product) {
        // 1. 先更新数据库
        db.save(product);
        // 2. 再删除缓存（而不是更新缓存）
        String key = "product:" + id;
        cache.delete(key);
    }
}

class ReadThroughCache:
    """缓存层封装了数据加载逻辑"""

    def __init__(self, redis_client, db_loader, default_ttl=1800):
        self.redis = redis_client
        self.db_loader = db_loader
        self.default_ttl = default_ttl

    def get(self, key: str) -> dict | None:
        # 1. 查缓存
        cached = self.redis.get(key)
        if cached is not None:
            return json.loads(cached)

        # 2. 缓存未命中，由缓存层自己去加载
        data = self.db_loader(key)
        if data is not None:
            self.redis.setex(key, self.default_ttl, json.dumps(data))
        return data


# 使用方式：应用代码不感知数据库
def product_loader(key: str) -> dict | None:
    product_id = key.split(":")[1]
    row = db.execute("SELECT * FROM product WHERE id = %s", (product_id,))
    return dict(row) if row else None

cache = ReadThroughCache(redis_client, product_loader)
product = cache.get("product:123")

public class WriteThroughCache {

    private final RedisTemplate<String, String> cache;
    private final JdbcTemplate db;

    @Transactional
    public void put(String key, Product product) {
        // 1. 写数据库
        db.update(
            "INSERT INTO product (id, name, price) VALUES (?, ?, ?) " +
            "ON DUPLICATE KEY UPDATE name=?, price=?",
            product.getId(), product.getName(), product.getPrice(),
            product.getName(), product.getPrice()
        );
        // 2. 写缓存（同步）
        String json = objectMapper.writeValueAsString(product);
        cache.opsForValue().set("product:" + product.getId(), json,
                                Duration.ofMinutes(30));
    }
}

import threading
import time
from collections import defaultdict

class WriteBehindCache:
    """写操作只写缓存，后台线程异步刷回数据库"""

    def __init__(self, redis_client, db_writer, flush_interval=5):
        self.redis = redis_client
        self.db_writer = db_writer
        self.dirty_keys = set()
        self.lock = threading.Lock()
        self.flush_interval = flush_interval
        self._start_flush_thread()

    def put(self, key: str, value: dict):
        # 只写缓存
        self.redis.set(key, json.dumps(value))
        with self.lock:
            self.dirty_keys.add(key)

    def _start_flush_thread(self):
        def flush_loop():
            while True:
                time.sleep(self.flush_interval)
                self._flush_to_db()
        t = threading.Thread(target=flush_loop, daemon=True)
        t.start()

    def _flush_to_db(self):
        with self.lock:
            keys = list(self.dirty_keys)
            self.dirty_keys.clear()

        if not keys:
            return

        batch = []
        for key in keys:
            value = self.redis.get(key)
            if value:
                batch.append((key, json.loads(value)))

        # 批量写数据库
        self.db_writer(batch)

package cache

import (
    "sync"
    "time"
)

type RefreshAheadCache struct {
    store       map[string]*entry
    mu          sync.RWMutex
    loader      func(key string) (interface{}, error)
    ttl         time.Duration
    refreshAt   float64 // 比如 0.8，表示在 80% 生命周期时触发刷新
    refreshing  map[string]bool
    refreshLock sync.Mutex
}

type entry struct {
    value     interface{}
    expiresAt time.Time
    createdAt time.Time
}

func (c *RefreshAheadCache) Get(key string) (interface{}, error) {
    c.mu.RLock()
    e, exists := c.store[key]
    c.mu.RUnlock()

    if !exists || time.Now().After(e.expiresAt) {
        // 缓存未命中或已过期，同步加载
        return c.loadAndSet(key)
    }

    // 检查是否需要提前刷新
    elapsed := time.Since(e.createdAt).Seconds()
    total := e.expiresAt.Sub(e.createdAt).Seconds()
    if elapsed/total >= c.refreshAt {
        // 触发异步刷新
        go c.asyncRefresh(key)
    }

    return e.value, nil
}

func (c *RefreshAheadCache) asyncRefresh(key string) {
    c.refreshLock.Lock()
    if c.refreshing[key] {
        c.refreshLock.Unlock()
        return
    }
    c.refreshing[key] = true
    c.refreshLock.Unlock()

    defer func() {
        c.refreshLock.Lock()
        delete(c.refreshing, key)
        c.refreshLock.Unlock()
    }()

    c.loadAndSet(key)
}

func (c *RefreshAheadCache) loadAndSet(key string) (interface{}, error) {
    value, err := c.loader(key)
    if err != nil {
        return nil, err
    }
    now := time.Now()
    c.mu.Lock()
    c.store[key] = &entry{
        value:     value,
        expiresAt: now.Add(c.ttl),
        createdAt: now,
    }
    c.mu.Unlock()
    return value, nil
}

# Redis 设置 TTL
SET product:123 '{"name":"iPhone","price":9999}' EX 1800
# 1800 秒（30 分钟）后自动过期

// 基于消息队列的缓存失效
@KafkaListener(topics = "product-changes")
public void onProductChange(ProductChangeEvent event) {
    String key = "product:" + event.getProductId();
    redisTemplate.delete(key);
    log.info("缓存已失效: {}", key);
}

# Debezium MySQL Connector 配置示例
name: product-cdc-connector
config:
  connector.class: io.debezium.connector.mysql.MySqlConnector
  database.hostname: mysql-master
  database.port: 3306
  database.user: cdc_user
  database.password: "${CDC_PASSWORD}"
  database.server.id: 1
  database.include.list: ecommerce
  table.include.list: ecommerce.product
  topic.prefix: cdc

class VersionedCache:
    def __init__(self, redis_client):
        self.redis = redis_client

    def get(self, key: str) -> dict | None:
        cached = self.redis.hgetall(f"vc:{key}")
        if not cached:
            return None

        # 获取当前版本
        current_version = self.redis.get(f"version:{key}")
        if current_version and cached.get("version") != current_version:
            # 版本不匹配，缓存已过期
            self.redis.delete(f"vc:{key}")
            return None

        return json.loads(cached["data"])

    def put(self, key: str, value: dict):
        version = str(int(time.time() * 1000))
        self.redis.hset(f"vc:{key}", mapping={
            "data": json.dumps(value),
            "version": version
        })
        self.redis.set(f"version:{key}", version)

    def invalidate(self, key: str):
        # 只需递增版本号，不用删数据
        new_version = str(int(time.time() * 1000))
        self.redis.set(f"version:{key}", new_version)

public Product getProduct(Long id) {
    String key = "product:" + id;
    // 尝试从缓存获取（包括空值标记）
    String cached = redisTemplate.opsForValue().get(key);
    if (cached != null) {
        if ("NULL".equals(cached)) {
            return null; // 命中空值缓存
        }
        return objectMapper.readValue(cached, Product.class);
    }

    Product product = productRepository.findById(id).orElse(null);
    if (product != null) {
        redisTemplate.opsForValue().set(key,
            objectMapper.writeValueAsString(product),
            30, TimeUnit.MINUTES);
    } else {
        // 缓存空值，TTL 设短一些
        redisTemplate.opsForValue().set(key, "NULL",
            5, TimeUnit.MINUTES);
    }
    return product;
}

from pybloom_live import BloomFilter

class BloomFilterProtectedCache:
    def __init__(self, redis_client, expected_items=1_000_000, fp_rate=0.001):
        self.redis = redis_client
        self.bloom = BloomFilter(capacity=expected_items, error_rate=fp_rate)

    def warm_up(self, all_valid_keys: list[str]):
        """启动时加载所有合法 key 到布隆过滤器"""
        for key in all_valid_keys:
            self.bloom.add(key)

    def get(self, key: str) -> dict | None:
        # 布隆过滤器判定：不存在则一定不存在
        if key not in self.bloom:
            return None

        # 可能存在，继续查缓存和数据库
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)

        data = self.load_from_db(key)
        if data:
            self.redis.setex(key, 1800, json.dumps(data))
        return data

public Product getProductWithMutex(Long id) {
    String key = "product:" + id;
    String lockKey = "lock:product:" + id;

    Product product = cacheGet(key);
    if (product != null) {
        return product;
    }

    // 尝试获取分布式锁
    boolean locked = redisTemplate.opsForValue()
        .setIfAbsent(lockKey, "1", 10, TimeUnit.SECONDS);

    if (locked) {
        try {
            // 双重检查：再查一次缓存，可能别的线程已经加载好了
            product = cacheGet(key);
            if (product != null) {
                return product;
            }
            // 从数据库加载
            product = productRepository.findById(id).orElse(null);
            if (product != null) {
                cacheSet(key, product, 30, TimeUnit.MINUTES);
            }
        } finally {
            redisTemplate.delete(lockKey);
        }
    } else {
        // 没拿到锁，短暂等待后重试
        try {
            Thread.sleep(50);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return getProductWithMutex(id);
    }
    return product;
}

import threading

class LogicalExpirationCache:
    def __init__(self, redis_client, db_loader):
        self.redis = redis_client
        self.db_loader = db_loader
        self.refresh_locks = {}

    def get(self, key: str) -> dict | None:
        raw = self.redis.get(key)
        if raw is None:
            return self._sync_load(key)

        entry = json.loads(raw)
        if time.time() < entry["expire_at"]:
            return entry["data"]

        # 逻辑过期，异步刷新，返回旧数据
        self._async_refresh(key)
        return entry["data"]

    def _async_refresh(self, key: str):
        if key in self.refresh_locks:
            return
        self.refresh_locks[key] = True

        def do_refresh():
            try:
                data = self.db_loader(key)
                if data:
                    entry = {
                        "data": data,
                        "expire_at": time.time() + 1800
                    }
                    self.redis.set(key, json.dumps(entry))
            finally:
                self.refresh_locks.pop(key, None)

        threading.Thread(target=do_refresh, daemon=True).start()

    def _sync_load(self, key: str) -> dict | None:
        data = self.db_loader(key)
        if data:
            entry = {
                "data": data,
                "expire_at": time.time() + 1800
            }
            self.redis.set(key, json.dumps(entry))
        return data

public void cacheWithJitter(String key, Object value, long baseTtlSeconds) {
    // 在基础 TTL 上加 0-300 秒的随机偏移
    long jitter = ThreadLocalRandom.current().nextLong(0, 300);
    long actualTtl = baseTtlSeconds + jitter;
    redisTemplate.opsForValue().set(key, value, actualTtl, TimeUnit.SECONDS);
}

@CircuitBreaker(name = "productService", fallbackMethod = "getProductFallback")
public Product getProduct(Long id) {
    // 正常查询逻辑
    return queryWithCache(id);
}

public Product getProductFallback(Long id, Throwable t) {
    // 降级：返回本地缓存的旧数据或默认值
    Product cached = localCache.getIfPresent(id);
    if (cached != null) {
        return cached;
    }
    return Product.defaultProduct();
}

import math
import random

def should_refresh_early(
    expire_at: float,
    now: float,
    beta: float = 1.0,
    recompute_time: float = 0.5
) -> bool:
    """
    XFetch 算法：概率性提前过期
    beta: 控制提前刷新的激进程度，值越大越激进
    recompute_time: 重新计算数据的预估耗时（秒）
    """
    ttl_remaining = expire_at - now
    # 到期前的随机提前量
    random_early = recompute_time * beta * math.log(random.random())
    # random_early 是负数，如果其绝对值大于剩余 TTL，则刷新
    return -random_early >= ttl_remaining


def get_with_xfetch(cache, key, loader, ttl=1800, beta=1.0):
    raw = cache.get(key)
    if raw:
        entry = json.loads(raw)
        now = time.time()
        if not should_refresh_early(entry["expire_at"], now, beta):
            return entry["data"]
        # 概率性触发异步刷新
        # 多数请求仍然返回旧数据，只有少数请求触发刷新
    data = loader(key)
    entry = {"data": data, "expire_at": time.time() + ttl}
    cache.set(key, json.dumps(entry))
    return data

// Caffeine 本地缓存配置
Cache<Long, Product> localCache = Caffeine.newBuilder()
    .maximumSize(10_000)           // 最多缓存 1 万个条目
    .expireAfterWrite(5, TimeUnit.MINUTES) // 写入 5 分钟后过期
    .recordStats()                  // 开启统计
    .build();

public class MultiLevelCacheService {

    private final Cache<String, Product> l1Cache; // Caffeine
    private final RedisTemplate<String, String> l2Cache; // Redis
    private final ProductRepository db;
    private final RedisMessageListenerContainer listenerContainer;

    public Product getProduct(Long id) {
        String key = "product:" + id;

        // L1: 进程内缓存
        Product product = l1Cache.getIfPresent(key);
        if (product != null) {
            return product;
        }

        // L2: Redis
        String json = l2Cache.opsForValue().get(key);
        if (json != null) {
            product = deserialize(json);
            l1Cache.put(key, product); // 回填 L1
            return product;
        }

        // L3: 数据库
        product = db.findById(id).orElse(null);
        if (product != null) {
            l2Cache.opsForValue().set(key, serialize(product),
                                      30, TimeUnit.MINUTES);
            l1Cache.put(key, product);
        }
        return product;
    }

    public void updateProduct(Long id, Product product) {
        db.save(product);
        String key = "product:" + id;
        l2Cache.delete(key);
        l1Cache.invalidate(key);
        // 通知其他节点失效 L1 缓存
        l2Cache.convertAndSend("cache-invalidation", key);
    }
}

// 发布失效消息
public void invalidateAllLevels(String key) {
    l2Cache.delete(key);
    l1Cache.invalidate(key);
    // 广播给所有节点
    redisTemplate.convertAndSend("cache:invalidation", key);
}

// 所有节点订阅失效消息
@Component
public class CacheInvalidationListener implements MessageListener {

    private final Cache<String, Product> l1Cache;

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String key = new String(message.getBody());
        l1Cache.invalidate(key);
    }
}

public void updateProductWithDoubleDelete(Long id, Product product) {
    String key = "product:" + id;

    // 第一次删除缓存
    redisTemplate.delete(key);

    // 更新数据库
    productRepository.save(product);

    // 延迟第二次删除（延迟时间 > 一次读请求的耗时）
    CompletableFuture.runAsync(() -> {
        try {
            Thread.sleep(500); // 延迟 500ms
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        redisTemplate.delete(key);
    });
}

# Canal 消费者示例（简化）
from canal.client import Client

def consume_binlog():
    client = Client()
    client.connect(host="canal-server", port=11111)
    client.subscribe(client_id="cache-invalidator", filter="ecommerce\\..*")

    while True:
        message = client.get(batch_size=100)
        for entry in message.entries:
            if entry.entry_type == "ROWDATA":
                for row in entry.row_changes:
                    table = entry.header.table_name
                    if table == "product":
                        product_id = get_column_value(row, "id")
                        redis_client.delete(f"product:{product_id}")
                        # 广播 L1 失效
                        redis_client.publish(
                            "cache:invalidation",
                            f"product:{product_id}"
                        )
        client.ack(message.id)

@Configuration
public class CacheConfig {

    @Bean
    public Cache<String, Product> productLocalCache() {
        return Caffeine.newBuilder()
            .maximumSize(50_000)
            .expireAfterWrite(2, TimeUnit.MINUTES)
            .refreshAfterWrite(90, TimeUnit.SECONDS)
            .build();
    }

    @Bean
    public Cache<String, ProductDetail> productDetailLocalCache() {
        return Caffeine.newBuilder()
            .maximumSize(20_000)
            .expireAfterWrite(5, TimeUnit.MINUTES)
            .build();
    }
}

public class ProductCacheService {

    private final Cache<String, Product> l1;
    private final StringRedisTemplate redis;
    private final ProductRepository db;

    public Product getProduct(Long id) {
        String key = "product:" + id;

        // L1
        Product product = l1.getIfPresent(key);
        if (product != null) {
            return product;
        }

        // L2: Redis（带防击穿逻辑）
        product = getFromRedisWithProtection(key, id);
        if (product != null) {
            l1.put(key, product);
        }
        return product;
    }

    private Product getFromRedisWithProtection(String key, Long id) {
        String json = redis.opsForValue().get(key);
        if (json != null) {
            return deserialize(json);
        }

        // 缓存未命中，用分布式锁防击穿
        String lockKey = "lock:" + key;
        Boolean locked = redis.opsForValue()
            .setIfAbsent(lockKey, "1", 5, TimeUnit.SECONDS);

        if (Boolean.TRUE.equals(locked)) {
            try {
                // 双重检查
                json = redis.opsForValue().get(key);
                if (json != null) {
                    return deserialize(json);
                }

                Product product = db.findById(id).orElse(null);
                if (product != null) {
                    // 防雪崩：TTL 加随机偏移
                    long ttl = 1800 + ThreadLocalRandom.current().nextLong(300);
                    redis.opsForValue().set(key,
                        serialize(product), ttl, TimeUnit.SECONDS);
                }
                return product;
            } finally {
                redis.delete(lockKey);
            }
        } else {
            // 没拿到锁，等待后重试（最多重试 3 次）
            sleep(50);
            String retryJson = redis.opsForValue().get(key);
            return retryJson != null ? deserialize(retryJson) : null;
        }
    }
}

@Component
public class ProductCdcConsumer {

    private final StringRedisTemplate redis;
    private final Cache<String, Product> l1Cache;

    @KafkaListener(topics = "cdc.ecommerce.product")
    public void onProductChange(String message) {
        CdcEvent event = parseEvent(message);
        String key = "product:" + event.getRowId();

        // 删除 L2 缓存
        redis.delete(key);

        // 广播 L1 失效
        redis.convertAndSend("cache:invalidation:product", key);

        // 如果是热点商品，主动预热
        if (isHotProduct(event.getRowId())) {
            Product product = loadFromDb(event.getRowId());
            if (product != null) {
                long ttl = 1800 + ThreadLocalRandom.current().nextLong(300);
                redis.opsForValue().set(key, serialize(product),
                    ttl, TimeUnit.SECONDS);
            }
        }
    }
}