第 26 章：综合项目实战：把 Redis 用进真实系统

@Service
public class ProductCacheService {

    private final StringRedisTemplate redis;
    private final ProductMapper productMapper;
    private final ObjectMapper objectMapper;
    private final Cache<Long, ProductBase> localCache;
    private final RateLimiter dbLimiter = RateLimiter.create(2000);

    public ProductCacheService(StringRedisTemplate redis,
                               ProductMapper productMapper,
                               ObjectMapper objectMapper) {
        this.redis = redis;
        this.productMapper = productMapper;
        this.objectMapper = objectMapper;
        this.localCache = Caffeine.newBuilder()
                .maximumSize(50_000)
                .expireAfterWrite(Duration.ofSeconds(3))
                .recordStats()
                .build();
    }

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

        ProductBase base = localCache.getIfPresent(id);
        if (base == null) {
            base = getJson("mall:product:base:" + id,
                    ProductBase.class, () -> productMapper.selectBase(id));
            if (base != null) {
                localCache.put(id, base);
            }
        }
        if (base == null) {
            return null;
        }

        ProductPrice price = getJson("mall:product:price:" + id,
                ProductPrice.class, () -> productMapper.selectPrice(id));
        ProductStat stat = getJson("mall:product:stat:" + id,
                ProductStat.class, () -> productMapper.selectStat(id));

        return ProductDetail.of(base, price, stat);
    }

    private <T> T getJson(String key, Class<T> type, Supplier<T> loader) {
        String value = redis.opsForValue().get(key);
        if (value != null) {
            if (CacheValues.NULL.equals(value)) {
                return null;
            }
            try {
                return objectMapper.readValue(value, type);
            } catch (JsonProcessingException e) {
                redis.delete(key);
            }
        }

        if (!dbLimiter.tryAcquire(100, TimeUnit.MILLISECONDS)) {
            throw new ServiceUnavailableException("database busy");
        }
        T loaded = loader.get();
        if (loaded == null) {
            redis.opsForValue().set(key, CacheValues.NULL, Duration.ofSeconds(60));
            return null;
        }
        redis.opsForValue().set(key, writeJson(loaded), randomTtl());
        return loaded;
    }

    private Duration randomTtl() {
        Duration base = Duration.ofMinutes(30);
        long jitter = ThreadLocalRandom.current().nextLong(0, 180);
        return base.plusSeconds(jitter);
    }
}

@Transactional
public void updateBase(ProductBase base) {
    productMapper.updateBase(base);

    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
        @Override
        public void afterCommit() {
            invalidateProduct(base.getId());
        }
    });
}

private void invalidateProduct(Long id) {
    List<String> keys = List.of(
            "mall:product:base:" + id,
            "mall:product:price:" + id,
            "mall:product:detail:" + id
    );
    try {
        redis.delete(keys);
    } catch (Exception e) {
        invalidationMapper.insert(new CacheInvalidation("mall:product", id, keys));
    }
}

@Scheduled(fixedDelay = 1000)
public void retryInvalidation() {
    List<CacheInvalidation> tasks = invalidationMapper.selectPending(100);
    for (CacheInvalidation task : tasks) {
        try {
            redis.delete(task.getKeys());
            invalidationMapper.markDone(task.getId());
        } catch (Exception e) {
            invalidationMapper.increaseRetry(task.getId());
        }
    }
}

@Service
public class SeckillService {

    private static final RedisScript<Long> SECKILL_SCRIPT = new DefaultRedisScript<>("""
            local status = redis.call('HGET', KEYS[1], 'status')
            if status ~= 'STARTED' then return -1 end
            if redis.call('SISMEMBER', KEYS[3], ARGV[1]) == 1 then return -3 end
            local stock = tonumber(redis.call('GET', KEYS[2]) or '0')
            if stock <= 0 then return -4 end
            redis.call('DECR', KEYS[2])
            redis.call('SADD', KEYS[3], ARGV[1])
            redis.call('SET', KEYS[4], ARGV[1], 'EX', tonumber(ARGV[3]))
            return 1
            """, Long.class);

    public SeckillResult submit(Long activityId, Long userId) {
        if (!activityStateHolder.isOpen(activityId)) {
            return SeckillResult.notStarted();
        }

        Long result = redisTemplate.execute(
                SECKILL_SCRIPT,
                List.of(
                        "seckill:activity:" + activityId,
                        "seckill:stock:" + activityId,
                        "seckill:user:" + activityId,
                        "seckill:order:" + activityId + ":" + userId
                ),
                String.valueOf(userId),
                String.valueOf(System.currentTimeMillis()),
                "900"
        );

        if (result == null || result <= 0) {
            return SeckillResult.fail(result);
        }

        seckillOrderProducer.send(SeckillOrderEvent.of(activityId, userId));
        return SeckillResult.success();
    }
}

public void compensate(Long activityId, Long userId) {
    redisTemplate.execute(releaseScript(),
            List.of("seckill:stock:" + activityId, "seckill:user:" + activityId),
            String.valueOf(userId));
}

public void publishPost(Post post) {
    postMapper.insert(post);
    redis.delete("post:content:" + post.getId());

    stringRedisTemplate.opsForZSet().add(
            "feed:outbox:" + post.getUserId(),
            String.valueOf(post.getId()),
            post.getPublishedAt().toEpochMilli()
    );

    if (isBigV(post.getUserId())) {
        return;
    }
    fanoutService.asyncFanout(post);
}

public void fanout(Post post) {
    String cursor = ScanOptions.SCAN_CURSOR_START;
    do {
        ScanOptions options = ScanOptions.scanOptions().count(1000).build();
        try (Cursor<String> scan = redis.opsForSet()
                .scan("feed:follower:" + post.getUserId(), options)) {
            while (scan.hasNext()) {
                String followerId = scan.next();
                stringRedisTemplate.opsForZSet().add(
                        "feed:inbox:" + followerId,
                        String.valueOf(post.getId()),
                        post.getPublishedAt().toEpochMilli()
                );
            }
            cursor = scan.getCursor();
        }
    } while (!"0".equals(cursor));
}

public void trimInbox(Long userId) {
    stringRedisTemplate.opsForZSet()
            .removeRange("feed:inbox:" + userId, 0, -1001);
}

public List<Post> readInbox(Long userId, long maxScore, int limit) {
    Set<String> ids = stringRedisTemplate.opsForZSet()
            .reverseRangeByScore("feed:inbox:" + userId,
                    Double.NEGATIVE_INFINITY, maxScore, 0, limit);
    return loadPosts(ids);
}

public List<Post> readTimeline(Long userId, long maxScore, int limit) {
    Set<String> following = stringRedisTemplate.opsForSet()
            .members("feed:following:" + userId);
    if (following == null || following.isEmpty()) {
        return List.of();
    }

    List<Post> merged = new ArrayList<>();
    for (String authorId : following) {
        Set<String> ids = stringRedisTemplate.opsForZSet()
                .reverseRangeByScore("feed:outbox:" + authorId,
                        Double.NEGATIVE_INFINITY, maxScore, 0, limit);
        merged.addAll(loadPosts(ids));
    }

    return merged.stream()
            .sorted(Comparator.comparingLong(Post::getPublishedAt).reversed())
            .limit(limit)
            .toList();
}

private List<Post> filterBlockedAuthors(Long userId, List<Post> posts) {
    Set<String> following = stringRedisTemplate.opsForSet()
            .members("feed:following:" + userId);
    return posts.stream()
            .filter(post -> following.contains(String.valueOf(post.getUserId())))
            .toList();
}

public void increaseSales(Long productId, int count) {
    String daily = "rank:product:daily:" + LocalDate.now().format(DateTimeFormatter.BASIC_ISO_DATE);
    String weekly = "rank:product:weekly:" + weekKey();
    String monthly = "rank:product:monthly:" + monthKey();

    redis.executePipelined(new SessionCallback<Void>() {
        @Override
        public Void execute(RedisOperations operations) {
            operations.opsForZSet().incrementScore(daily, String.valueOf(productId), count);
            operations.opsForZSet().incrementScore(weekly, String.valueOf(productId), count);
            operations.opsForZSet().incrementScore(monthly, String.valueOf(productId), count);
            return null;
        }
    });
}

public List<RankItem> topN(String rankKey, int n) {
    Set<ZSetOperations.TypedTuple<String>> tuples =
            stringRedisTemplate.opsForZSet().reverseRangeWithScores(rankKey, 0, n - 1);
    if (tuples == null) {
        return List.of();
    }

    List<RankItem> items = new ArrayList<>();
    int rank = 1;
    for (ZSetOperations.TypedTuple<String> tuple : tuples) {
        items.add(new RankItem(rank++, Long.valueOf(tuple.getValue()), tuple.getScore()));
    }
    return items;
}

public RankItem getUserRank(String rankKey, Long userId) {
    Long rank = stringRedisTemplate.opsForZSet()
            .reverseRank(rankKey, String.valueOf(userId));
    Double score = stringRedisTemplate.opsForZSet()
            .score(rankKey, String.valueOf(userId));
    if (rank == null || score == null) {
        return null;
    }
    return new RankItem(rank + 1, userId, score);
}

List<RankItem> items = loadCandidates(rankKey, 500);
return items.stream()
        .sorted(Comparator.comparingDouble(RankItem::score).reversed()
                .thenComparingLong(RankItem::achievedAt))
        .limit(100)
        .toList();

@Scheduled(cron = "0 10 0 * * ?")
public void archiveDailyRank() {
    String key = "rank:product:daily:" + LocalDate.now().minusDays(1)
            .format(DateTimeFormatter.BASIC_ISO_DATE);
    Set<ZSetOperations.TypedTuple<String>> tuples =
            stringRedisTemplate.opsForZSet().reverseRangeWithScores(key, 0, 999);

    if (tuples == null || tuples.isEmpty()) {
        return;
    }
    rankHistoryMapper.batchInsert(buildRows(key, tuples));
    stringRedisTemplate.expire(key, Duration.ofDays(7));
}

@Transactional
public void createOrder(CreateOrderRequest request) {
    Order order = orderDomainService.create(request);
    orderMapper.insert(order);

    stringRedisTemplate.opsForZSet().add(
            "order:timeout:zset",
            String.valueOf(order.getId()),
            order.getCreateTime().plusMinutes(15).toEpochMilli()
    );
}

@Scheduled(fixedDelay = 1000)
public void closeTimeoutOrders() {
    long now = System.currentTimeMillis();
    Set<String> orderIds = stringRedisTemplate.opsForZSet()
            .rangeByScore("order:timeout:zset", 0, now, 0, 99);

    if (orderIds == null || orderIds.isEmpty()) {
        return;
    }
    orderIds.forEach(orderId -> closeIfTimeout(Long.valueOf(orderId)));
}

public void closeIfTimeout(Long orderId) {
    String lockKey = "order:lock:close:" + orderId;
    String token = UUID.randomUUID().toString();
    Boolean locked = stringRedisTemplate.opsForValue()
            .setIfAbsent(lockKey, token, Duration.ofSeconds(10));

    if (!Boolean.TRUE.equals(locked)) {
        return;
    }

    try {
        int updated = orderMapper.closeIfUnpaid(orderId,
                OrderStatus.UNPAID, OrderStatus.CLOSED);
        if (updated > 0) {
            inventoryService.release(orderId);
            couponService.release(orderId);
            eventPublisher.publish(OrderClosedEvent.of(orderId));
        }
        stringRedisTemplate.opsForZSet()
                .remove("order:timeout:zset", String.valueOf(orderId));
    } finally {
        releaseLock(lockKey, token);
    }
}

@Scheduled(cron = "0 */5 * * * ?")
public void scanDatabaseTimeoutOrders() {
    List<Order> orders = orderMapper.selectTimeoutUnpaid(500);
    orders.forEach(order -> closeIfTimeout(order.getId()));
}