第 25 章：缓存架构设计：从会用缓存到设计缓存

public Duration cacheTtl(Duration base) {
    long jitter = ThreadLocalRandom.current()
            .nextLong(0, Math.max(1, base.toSeconds() / 10));
    return base.plusSeconds(jitter);
}

public Product getProduct(Long id) {
    String key = "mall:product:base:" + id;
    String cached = redis.opsForValue().get(key);
    if (cached != null) {
        if ("NULL".equals(cached)) {
            return null;
        }
        return objectMapper.readValue(cached, Product.class);
    }

    Product product = productMapper.selectById(id);
    if (product == null) {
        redis.opsForValue().set(key, "NULL", Duration.ofSeconds(60));
        return null;
    }
    redis.opsForValue().set(key, writeJson(product), cacheTtl(Duration.ofMinutes(30)));
    return product;
}

@Transactional
public void updateProduct(Product product) {
    productMapper.updateById(product);
    redis.delete("mall:product:base:" + product.getId());
}

public void increaseView(Long productId) {
    redis.opsForZSet().incrementScore(
            "mall:product:view:buffer", String.valueOf(productId), 1
    );
}

@Scheduled(fixedDelay = 5000)
public void flushViewBuffer() {
    String key = "mall:product:view:buffer";
    Set<ZSetOperations.TypedTuple<String>> batch =
            redis.opsForZSet().rangeWithScores(key, 0, 499);
    if (batch == null || batch.isEmpty()) {
        return;
    }

    List<ViewUpdate> updates = batch.stream()
            .map(tuple -> new ViewUpdate(
                    Long.valueOf(tuple.getValue()),
                    tuple.getScore().longValue()))
            .toList();

    for (ZSetOperations.TypedTuple<String> tuple : batch) {
        redis.opsForZSet().remove(key, tuple.getValue());
    }
    viewMapper.batchIncrease(updates);
}

if (id == null || id <= 0) {
    throw new BadRequestException("invalid product id");
}

if (product == null) {
    redis.opsForValue().set(key, "", Duration.ofSeconds(60));
    return Product.empty(id);
}

public boolean mayExist(Long id) {
    return bloomFilter.mightContain("mall:product:" + id);
}

public Product getProductWithLock(Long id) {
    String key = "mall:product:base:" + id;
    Product cached = readCache(key);
    if (cached != null) {
        return cached;
    }

    String lockKey = "lock:load:product:" + id;
    String token = UUID.randomUUID().toString();
    Boolean locked = redis.opsForValue()
            .setIfAbsent(lockKey, token, Duration.ofSeconds(5));

    if (Boolean.TRUE.equals(locked)) {
        try {
            cached = readCache(key);
            if (cached != null) {
                return cached;
            }
            Product product = productMapper.selectById(id);
            writeCache(key, product, Duration.ofMinutes(30));
            return product;
        } finally {
            release(lockKey, token);
        }
    }

    sleepQuietly(30);
    return getProduct(id);
}

public Product safeGetProduct(Long id) {
    Product cached = localCache.getIfPresent(id);
    if (cached != null) {
        return cached;
    }

    try {
        cached = redisGetProduct(id);
        if (cached != null) {
            localCache.put(id, cached);
            return cached;
        }
    } catch (RedisConnectionException e) {
        log.warn("Redis unavailable, product id={}", id, e);
    }

    if (!databaseLimiter.tryAcquire()) {
        throw new ServiceUnavailableException("cache and db are busy");
    }
    return productMapper.selectById(id);
}

@Transactional
public void update(Product product) {
    String key = "mall:product:base:" + product.getId();
    redis.delete(key);
    productMapper.updateById(product);

    delayQueue.send(new CacheInvalidation(key), Duration.ofMillis(500));
}

@KafkaListener(topics = "mysql.mall.product")
public void onProductChange(ConsumerRecord<String, String> record) {
    ProductChange event = objectMapper.readValue(record.value(), ProductChange.class);
    if (event.after() != null) {
        String key = "mall:product:base:" + event.after().getId();
        invalidationGuard.deleteIfVersionMatch(key, event.after().getVersion());
    }
}

private final Cache<Long, Product> localCache = Caffeine.newBuilder()
        .maximumSize(20_000)
        .expireAfterWrite(Duration.ofSeconds(5))
        .recordStats()
        .build();

redis.convertAndSend("cache:invalidate", "mall:product:base:10001");

public void onMessage(String message) {
    localCache.invalidate(extractId(message));
}

String key = "mall:product:base:" + id + ":" + ThreadLocalRandom.current().nextInt(16);
Product product = readFromRedis(key);
if (product == null) {
    product = loadAndFillAllCopies(id);
}

@Scheduled(cron = "0 30 23 * * ?")
public void warmUpTomorrowTopProducts() {
    List<Long> ids = productMapper.selectTomorrowTop(50_000);
    for (int i = 0; i < ids.size(); i += 500) {
        List<Long> batch = ids.subList(i, Math.min(i + 500, ids.size()));
        cacheLoader.asyncWarmUp(batch);
        sleepQuietly(200);
    }
}

RateLimiter dbLimiter = RateLimiter.create(2000); // 每秒最多 2000 次查库

public Product loadFromDatabase(Long id) {
    if (!dbLimiter.tryAcquire(50, TimeUnit.MILLISECONDS)) {
        throw new ServiceUnavailableException("database limiter rejected");
    }
    return productMapper.selectById(id);
}

public <T> T getWithMetrics(String key, Class<T> type, Supplier<T> loader) {
    long start = System.nanoTime();
    String value = redis.get(key);
    long redisCost = System.nanoTime() - start;

    if (value != null) {
        metrics.recordHit(prefix(key), redisCost);
        return deserialize(value, type);
    }

    metrics.recordMiss(prefix(key), redisCost);
    long dbStart = System.nanoTime();
    T loaded = loader.get();
    metrics.recordLoad(prefix(key), System.nanoTime() - dbStart);
    return loaded;
}

public Product getProduct(Long id) {
    validateId(id);

    Product product = localCache.getIfPresent(id);
    if (product != null) {
        return product;
    }

    try {
        product = redisCache.getProduct(id);
        if (product != null) {
            localCache.put(id, product);
            return product;
        }
    } catch (CacheUnavailableException e) {
        metrics.redisFailure("product");
    }

    if (!loadLimiter.tryAcquire()) {
        return fallbackProduct(id);
    }

    product = productMapper.selectById(id);
    if (product == null) {
        redisCache.putEmpty(id, Duration.ofSeconds(60));
        return null;
    }

    redisCache.putProduct(id, product, randomTtl(Duration.ofMinutes(30)));
    localCache.put(id, product);
    return product;
}