黑马点评 (heimadianping) 技术亮点

这是一个仿大众点评的实战项目，涵盖了 Redis 缓存、分布式锁、秒杀系统、Feed 流等核心技术。本文将深入分析项目中的关键技术实现和设计思路。

1. Redis 缓存策略

1.1 缓存穿透的解决方案

缓存穿透是指查询一个一定不存在的数据，由于缓存不命中，每次都要去数据库查询，但数据库也没有这条数据，导致永远不会写入缓存。

项目采用了缓存空值的方案来解决：

public <T, ID> T queryWithPassThrough(String keyPrefix, ID id, Class<T> type,
                                        Function<ID, T> dbFallback, Long time, TimeUnit unit) {
    String key = keyPrefix + id;
    String json = stringRedisTemplate.opsForValue().get(key);

    // 命中缓存，直接返回
    if (StrUtil.isNotBlank(json)) {
        return JSONUtil.toBean(json, type);
    }

    // 命中空值（之前缓存的""），直接返回null
    if (json != null) {
        return null;
    }

    // 未命中缓存，查询数据库
    T t = dbFallback.apply(id);

    if (t == null) {
        // 数据库也没有，缓存空值，设置较短的TTL
        this.set(key, "", time, unit);
        return null;
    }

    // 正常缓存
    this.set(key, t, time, unit);
    return t;
}

设计要点：

• 使用 StrUtil.isNotBlank(json) 区分"未缓存"和"缓存了空字符串"
• 空值的 TTL 设置较短（2分钟），避免占用过多内存
• 通过泛型和函数式接口 Function<ID, T> dbFallback 实现通用性

1.2 缓存击穿的解决方案

缓存击穿是指热点key在某个时间点过期，恰好在这个时间点对这个key有大量的并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。

项目提供了两种方案：

方案一：互斥锁

public Shop queryWithPassThrough(Long id) {
    String key = CACHE_SHOP_KEY + id;
    String shopJson = stringRedisTemplate.opsForValue().get(key);

    if (StrUtil.isNotBlank(shopJson)) {
        return JSONUtil.toBean(shopJson, Shop.class);
    }

    // 获取互斥锁
    String lockKey = LOCK_SHOP_KEY + id;
    boolean isLocked = tryLock(lockKey);

    if (!isLocked) {
        // 获取锁失败，休眠后重试
        Thread.sleep(50);
        return queryWithPassThrough(id); // 递归重试
    }

    try {
        // 双重检验
        shopJson = stringRedisTemplate.opsForValue().get(key);
        if (StrUtil.isNotBlank(shopJson)) {
            return JSONUtil.toBean(shopJson, Shop.class);
        }

        // 查询数据库并缓存
        Shop shop = this.getById(id);
        // ...
        return shop;
    } finally {
        unlock(lockKey);
    }
}

方案二：逻辑过期（推荐）

public <R, ID> R queryWithLogicalExpire(String keyPrefix, ID id, Class<R> type,
                                         Function<ID, R> dbFallback, Long time, TimeUnit unit) {
    String key = keyPrefix + id;
    String shopJson = stringRedisTemplate.opsForValue().get(key);

    if (StrUtil.isBlank(shopJson)) {
        return null; // 缓存未命中，说明不是热点数据
    }

    RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);
    LocalDateTime expireTime = redisData.getExpireTime();
    R r = JSONUtil.toBean((JSONObject) redisData.getData(), type);

    // 逻辑过期时间未到，直接返回
    if (expireTime.isAfter(LocalDateTime.now())) {
        return r;
    }

    // 已过期，尝试获取锁重建缓存
    String lockKey = LOCK_SHOP_KEY + id;
    boolean isLock = tryLock(lockKey);

    if (isLock) {
        // 获取锁成功，开启独立线程重建缓存
        CACHE_REBUILD_EXECUTOR.submit(() -> {
            try {
                R rl = dbFallback.apply(id);
                this.setWithLogicalExpire(key, rl, time, unit);
            } finally {
                unlock(lockKey);
            }
        });
    }

    // 返回旧数据（过期但仍可用）
    return r;
}

RedisData 结构：

@Data
public class RedisData {
    private LocalDateTime expireTime;  // 逻辑过期时间
    private Object data;               // 实际数据
}

逻辑过期的优势：

• 不阻塞用户请求，返回旧数据的同时异步更新
• 不需要真正的Redis TTL，通过程序判断过期
• 适合读多写少的热点数据场景

1.3 缓存雪崩的预防

项目通过以下策略预防缓存雪崩：

• 给缓存TTL添加随机值，避免同时过期
• 使用逻辑过期方案，热点数据永不过期
• Redis集群保证高可用

2. 秒杀功能实现

2.1 整体架构

秒杀功能采用了Lua脚本 + Redis Stream + 异步下单的架构：

用户请求 -> Lua脚本(原子操作) -> Redis Stream -> 异步线程消费 -> 数据库下单

2.2 Lua 脚本保证原子性

seckill.lua 是整个秒杀的核心：

-- 获取参数
local voucherId = ARGV[1]
local userId = ARGV[2]
local orderId = ARGV[3]

-- 构造Redis键
local stockKey = "seckill:stock:" .. voucherId
local orderKey = "seckill:order:" .. voucherId

-- 1. 检查库存是否充足
if (tonumber(redis.call("get", stockKey)) <= 0) then
    return 1  -- 库存不足
end

-- 2. 检查用户是否已下单（一人一单）
if (redis.call("sismember", orderKey, userId) == 1) then
    return 2  -- 重复购买
end

-- 3. 扣减库存
redis.call("incrby", stockKey, -1)

-- 4. 记录用户购买信息
redis.call("sadd", orderKey, userId)

-- 5. 将订单信息发送到Stream队列
redis.call("xadd", "stream.orders", "*", "userId", userId, "voucherId", voucherId, "id", orderId)

-- 返回0表示秒杀成功
return 0

Lua脚本的优势：

• 原子性：整个脚本在Redis中原子执行，不会出现并发问题
• 减少网络开销：一次网络请求完成多个操作
• 提前过滤：在Redis层面就过滤掉无效请求，减轻数据库压力

2.3 Java 端调用

private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

static {
    SECKILL_SCRIPT = new DefaultRedisScript<>();
    SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
    SECKILL_SCRIPT.setResultType(Long.class);
}

@Override
public Result seckillVoucher(Long voucherId) {
    Long userId = UserHolder.getUser().getId();
    long orderId = redisIdWorker.nextId(ORDER);

    // 执行Lua脚本
    Long result = stringRedisTemplate.execute(
            SECKILL_SCRIPT,
            Collections.emptyList(),
            voucherId.toString(), userId.toString(), String.valueOf(orderId)
    );

    int r = result.intValue();
    if (r != 0) {
        return Result.fail(r == 1 ? "库存不足" : "不能重复下单");
    }

    // 获取代理对象（用于后续事务调用）
    proxy = (IVoucherOrderService) AopContext.currentProxy();

    return Result.ok(orderId);
}

2.4 Redis Stream 异步消费

项目使用 Redis Stream 作为消息队列，异步处理订单：

@PostConstruct
private void init() {
    SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderTask());
}

private class VoucherOrderTask implements Runnable {
    private static final String QUEUE_NAME = "stream.orders";

    @Override
    public void run() {
        while (true) {
            try {
                // 从Stream中读取消息
                List<MapRecord<String, Object, Object>> mapRecordList =
                    stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                        StreamOffset.create(QUEUE_NAME, ReadOffset.lastConsumed())
                    );

                if (mapRecordList == null || mapRecordList.isEmpty()) {
                    continue;
                }

                MapRecord<String, Object, Object> record = mapRecordList.get(0);
                Map<Object, Object> values = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);

                // 处理订单
                handleVoucherOrder(voucherOrder);

                // ACK确认
                stringRedisTemplate.opsForStream().acknowledge(QUEUE_NAME, "g1", record.getId());
            } catch (Exception e) {
                log.error("处理订单异常:", e);
                handlePendingList(); // 处理pending-list
            }
        }
    }
}

为什么选择 Redis Stream 而不是 BlockingQueue？

代码中保留了早期使用 BlockingQueue 的注释，从 BlockingQueue 升级到 Redis Stream 的原因是：

• 持久化：Stream 数据会持久化，服务重启不会丢失
• 消费者组：支持消费者组，可以实现消息的可靠消费
• ACK机制：支持消息确认，未确认的消息可以重新消费

2.5 分布式锁保证一人一单

private void handleVoucherOrder(VoucherOrder voucherOrder) {
    Long userId = voucherOrder.getUserId();

    // 使用Redisson分布式锁
    RLock lock = redissonClient.getLock(ORDER + userId);
    boolean tryLock = lock.tryLock();

    if (!tryLock) {
        return;
    }

    try {
        proxy.createVoucherOrder(voucherOrder);
    } finally {
        lock.unlock();
    }
}

@Transactional(rollbackFor = Exception.class)
public void createVoucherOrder(VoucherOrder voucherOrder) {
    Long userId = voucherOrder.getUserId();
    Long voucherId = voucherOrder.getVoucherId();

    // 一人一单校验
    Integer count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
    if (count > 0) {
        log.error("用户已经购买过一次");
        return;
    }

    // 乐观锁扣减库存
    boolean isUpdate = seckillVoucherService.update(new LambdaUpdateWrapper<SeckillVoucher>()
            .setSql("stock = stock - 1")
            .eq(SeckillVoucher::getVoucherId, voucherId)
            .gt(SeckillVoucher::getStock, 0));

    if (!isUpdate) {
        log.error("库存不足");
        return;
    }

    save(voucherOrder);
}

3. Feed 流实现（推模式）

3.1 发布笔记时推送给粉丝

项目采用推模式实现Feed流：用户发布笔记时，主动推送到所有粉丝的收件箱。

@Override
public Result saveBlog(Blog blog) {
    UserDTO user = UserHolder.getUser();
    Long userId = user.getId();
    blog.setUserId(userId);

    // 保存笔记到数据库
    boolean save = this.save(blog);

    if (BooleanUtil.isTrue(save)) {
        // 查询当前用户的所有粉丝
        List<Follow> fans = followService.lambdaQuery()
                .eq(Follow::getFollowUserId, userId)
                .list();

        // 将新笔记推送到每个粉丝的收件箱
        for (Follow fan : fans) {
            Long fanId = fan.getUserId();
            String key = FEED + fanId;

            // 使用ZSet存储，score为时间戳
            stringRedisTemplate.opsForZSet().add(
                key, blog.getId().toString(), System.currentTimeMillis()
            );
        }
    }

    return save ? Result.ok(blog.getId()) : Result.fail("发布笔记失败！");
}

3.2 查询关注的人的笔记（滚动分页）

@Override
public Result queryBlogOfFollow(Long max, Integer offset) {
    UserDTO user = UserHolder.getUser();
    Long userId = user.getId();
    String key = FEED + userId;

    // 从ZSet中获取数据，按时间倒序，支持滚动分页
    Set<ZSetOperations.TypedTuple<String>> typedTuples =
        stringRedisTemplate.opsForZSet().reverseRangeByScoreWithScores(
            key, 0, max, offset, 5  // 每次获取5条
        );

    if (typedTuples == null || typedTuples.isEmpty()) {
        return Result.ok();
    }

    List<Long> ids = new ArrayList<>(typedTuples.size());
    long minTime = max;
    int count = 0;

    for (ZSetOperations.TypedTuple<String> tuple : typedTuples) {
        if (tuple == null || tuple.getValue() == null || tuple.getScore() == null) {
            continue;
        }
        ids.add(Long.valueOf(tuple.getValue()));
        long time = tuple.getScore().longValue();

        // 记录最小时间和相同时间的数量（用于下次分页）
        if (time < minTime) {
            minTime = time;
            count = 1;
        } else if (time == minTime) {
            count += 1;
        }
    }

    // 根据ID查询笔记详情，保持顺序
    String idsStr = StrUtil.join(",", ids);
    List<Blog> blogs = query()
            .in("id", ids)
            .last("order by field(id," + idsStr + ")")
            .list();

    // 构建滚动分页结果
    ScrollResult scrollResult = new ScrollResult();
    scrollResult.setList(blogs);
    scrollResult.setOffset(count);
    scrollResult.setMinTime(minTime);

    return Result.ok(scrollResult);
}

3.3 Feed 流设计要点

为什么选择 ZSet？

• 天然支持按时间排序（score 为时间戳）
• 支持范围查询，方便分页
• 自动去重

滚动分页的实现：

• max：上一次查询的最小时间戳
• offset：与最小时间戳相同的数量（用于跳过）
• 每次查询返回 5 条数据

推模式的优缺点：

• 优点：读取速度快，用户查询自己的收件箱即可
• 缺点：粉丝多的用户发布时，写入压力大
• 适合：读多写少的场景

4. 登录认证（Token + Redis）

4.1 登录流程

@Override
public Result login(LoginFormDTO loginForm, HttpSession session) {
    String phone = loginForm.getPhone();

    // 校验验证码
    String code = loginForm.getCode();
    String cacheCode = stringRedisTemplate.opsForValue().get(RedisConstants.LOGIN_CODE_KEY + phone);

    if (cacheCode == null || !cacheCode.equals(code)) {
        return Result.fail("验证码错误！");
    }

    // 查询或创建用户
    User user = query().eq("phone", phone).one();
    if (user == null) {
        user = createUserWithPhone(phone);
        this.save(user);
    }

    // 生成Token
    String token = UUID.randomUUID().toString(true);

    // 将用户信息存入Redis Hash
    UserDTO userDTO = BeanUtil.copyProperties(user, UserDTO.class);
    Map<String, Object> userMap = BeanUtil.beanToMap(userDTO);

    // 转换为String类型（Redis Hash要求）
    Map<String, String> stringUserMap = new HashMap<>();
    for (Map.Entry<String, Object> entry : userMap.entrySet()) {
        stringUserMap.put(entry.getKey(),
            entry.getValue() != null ? entry.getValue().toString() : null);
    }

    // 存储到Redis，设置TTL
    stringRedisTemplate.opsForHash().putAll(LOGIN_USER_KEY + token, stringUserMap);
    stringRedisTemplate.expire(LOGIN_USER_KEY + token, LOGIN_USER_TTL, TimeUnit.MINUTES);

    return Result.ok(token);
}

4.2 双拦截器设计

项目使用两个拦截器实现认证：

RefreshTokenInterceptor（order=0）：拦截所有请求

public class RefreshTokenInterceptor implements HandlerInterceptor {

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response,
                             Object handler) throws Exception {
        // 1. 获取token
        String token = request.getHeader("Authorization");
        if (StrUtil.isBlank(token)) {
            return true;  // 放行，交给下一个拦截器判断
        }

        // 2. 从Redis获取用户信息
        Map<Object, Object> userMap = stringRedisTemplate.opsForHash()
            .entries(RedisConstants.LOGIN_USER_KEY + token);

        if (userMap.isEmpty()) {
            return true;
        }

        // 3. 存储用户到ThreadLocal
        UserDTO userDTO = BeanUtil.fillBeanWithMap(userMap, new UserDTO(), false);
        UserHolder.saveUser(userDTO);

        // 4. 刷新Token有效期
        stringRedisTemplate.expire(RedisConstants.LOGIN_USER_KEY + token,
            RedisConstants.LOGIN_USER_TTL, TimeUnit.MINUTES);

        return true;
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response,
                                Object handler, Exception ex) throws Exception {
        UserHolder.removeUser();  // 清理ThreadLocal
    }
}

LoginInterceptor（order=1）：拦截需要登录的请求

public class LoginInterceptor implements HandlerInterceptor {

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response,
                             Object handler) throws Exception {
        // 判断ThreadLocal中是否有用户
        UserDTO user = UserHolder.getUser();
        if (user == null) {
            response.setStatus(HttpServletResponse.SC_UNAUTHORIZED);
            return false;
        }
        return true;
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response,
                                Object handler, Exception ex) throws Exception {
        UserHolder.removeUser();
    }
}

拦截器配置：

@Override
public void addInterceptors(InterceptorRegistry registry) {
    // 第一个拦截器：刷新Token，拦截所有请求
    registry.addInterceptor(new RefreshTokenInterceptor(stringRedisTemplate))
            .addPathPatterns("/**")
            .order(0);

    // 第二个拦截器：登录校验，排除公开接口
    registry.addInterceptor(new LoginInterceptor())
            .excludePathPatterns(
                    "/user/code",
                    "/user/login",
                    "/blog/hot",
                    "/shop/**",
                    "/shop-type/**",
                    "/upload/**",
                    "/voucher/**"
            )
            .order(1);
}

设计优势：

• 职责分离：刷新Token和登录校验分开，逻辑清晰
• 统一刷新：无论是否需要登录，只要有Token就刷新有效期
• ThreadLocal传递：通过ThreadLocal在拦截器和Controller之间传递用户信息

4.3 用户签到（BitMap）

项目还实现了基于Redis BitMap的用户签到功能：

@Override
public Result sign() {
    Long userId = UserHolder.getUser().getId();
    LocalDateTime now = LocalDateTime.now();

    // 拼接key：sign:userId:yyyyMM
    String keySuffix = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));
    String key = "sign:" + userId + keySuffix;
    int day = now.getDayOfMonth();

    // 检查是否已签到
    if (BooleanUtil.isTrue(stringRedisTemplate.opsForValue().getBit(key, day - 1))) {
        return Result.fail("请勿重复签到！");
    }

    // 设置签到位
    stringRedisTemplate.opsForValue().setBit(key, day - 1, true);
    return Result.ok();
}

@Override
public Result signCount() {
    Long userId = UserHolder.getUser().getId();
    LocalDateTime now = LocalDateTime.now();

    String keySuffix = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));
    String key = "sign:" + userId + keySuffix;
    int day = now.getDayOfMonth();

    // 使用bitfield获取从第0位开始的day位
    List<Long> result = stringRedisTemplate.opsForValue().bitField(
            key,
            BitFieldSubCommands.create()
                    .get(BitFieldSubCommands.BitFieldType.unsigned(day))
                    .valueAt(0)
    );

    if (result == null || result.isEmpty()) {
        return Result.ok(0);
    }

    Long num = result.get(0);
    if (num == null || num == 0L) {
        return Result.ok(0);
    }

    // 统计连续签到天数
    int ans = 0;
    while ((num & 1) != 0) {
        ans++;
        num >>>= 1;
    }

    return Result.ok(ans);
}

BitMap的优势：

• 极其节省内存：1亿用户一个月的签到数据只需约12MB
• 支持位运算：可以快速统计连续签到、月签到次数等

5. 分布式ID生成器

项目使用自研的基于Redis的ID生成器：

@Component
@RequiredArgsConstructor
public class RedisIdWorker {
    private static final long BEGIN_TIMESTAMP = 1640995200L;  // 2022-01-01 00:00:00
    private static final long COUNT_BITS = 32;

    private final StringRedisTemplate stringRedisTemplate;

    public long nextId(String keyPrefix) {
        // 1. 生成时间戳（秒级）
        LocalDateTime now = LocalDateTime.now();
        long nowSecond = now.toEpochSecond(ZoneOffset.UTC);
        long timestamp = nowSecond - BEGIN_TIMESTAMP;

        // 2. 生成序列号（每天从0开始）
        String date = now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));
        Long countBox = stringRedisTemplate.opsForValue().increment("icr:" + keyPrefix + ":" + date);
        long count = countBox != null ? countBox : 1;

        // 3. 拼接：时间戳左移32位 + 序列号
        return timestamp << COUNT_BITS | count;
    }
}

ID结构：

|<-- 32位时间戳 -->|<-- 32位序列号 -->|

特点：

• 全局唯一：时间戳 + 序列号保证唯一性
• 趋势递增：高位是时间戳，整体递增
• 每天重置序列号：避免序列号无限增长

6. 关注功能（Set 实现共同关注）

@Override
public Result followCommons(Long id) {
    Long userId = UserHolder.getUser().getId();

    String keyCurUser = "follow:" + userId;
    String keyOtherUser = "follow:" + id;

    // 求交集：共同关注的人
    Set<String> intersect = stringRedisTemplate.opsForSet().intersect(keyCurUser, keyOtherUser);

    if (intersect == null || intersect.isEmpty()) {
        return Result.ok(Collections.emptyList());
    }

    // 移除自己
    intersect.remove(String.valueOf(userId));

    // 查询用户信息
    List<Long> ids = intersect.stream()
            .map(Long::valueOf)
            .collect(Collectors.toList());

    List<UserDTO> userDTOS = userService.listByIds(ids).stream()
            .map(user -> BeanUtil.copyProperties(user, UserDTO.class))
            .toList();

    return Result.ok(userDTOS);
}

Set的优势：

• 天然支持交集、并集、差集操作
• 共同关注 = 两个用户关注列表的交集

7. 附近商铺（Redis GEO）

@Override
public Result queryShopByType(Integer typeId, Integer current, Double x, Double y) {
    if (x == null || y == null) {
        // 无坐标时按类型分页查询
        Page<Shop> page = this.query()
                .eq("type_id", typeId)
                .page(new Page<>(current, SystemConstants.DEFAULT_PAGE_SIZE));
        return Result.ok(page.getRecords());
    }

    // 使用Redis GEO搜索附近5000米内的商铺
    String key = "shop:geo:" + typeId;
    int from = (current - 1) * SystemConstants.DEFAULT_PAGE_SIZE;
    int end = current * SystemConstants.DEFAULT_PAGE_SIZE;

    GeoResults<RedisGeoCommands.GeoLocation<String>> results =
        stringRedisTemplate.opsForGeo().radius(
            key,
            new Circle(new Point(x, y), new Distance(SEARCH_RADIUS)),
            RedisGeoCommands.GeoRadiusCommandArgs.newGeoRadiusArgs()
                .includeDistance()
                .limit(end)
        );

    // 解析结果，分页处理
    List<GeoResult<RedisGeoCommands.GeoLocation<String>>> list = results.getContent();

    List<Long> ids = new ArrayList<>();
    Map<String, Distance> distanceMap = new HashMap<>();

    list.stream().skip(from).forEach(result -> {
        String shopIdStr = result.getContent().getName();
        ids.add(Long.valueOf(shopIdStr));
        distanceMap.put(shopIdStr, result.getDistance());
    });

    // 查询商铺信息并保持顺序
    String idsStr = StrUtil.join(",", ids);
    List<Shop> shops = query()
        .in("id", ids)
        .last("ORDER BY FIELD(id," + idsStr + ")")
        .list();

    // 设置距离
    for (Shop shop : shops) {
        shop.setDistance(distanceMap.get(shop.getId().toString()).getValue());
    }

    return Result.ok(shops);
}

GEO的应用场景：

• 附近的人/店铺
• 距离排序
• 地理围栏

8. 总结

核心技术栈

• Redis：缓存、分布式锁、消息队列、BitMap、GEO、Set、ZSet
• Redisson：分布式锁的成熟实现
• Lua脚本：保证复杂操作的原子性
• Spring Boot：拦截器、事务管理

设计亮点

1. 分层缓存：本地缓存 + Redis缓存 + 数据库
2. 异步处理：秒杀请求异步下单，削峰填谷
3. 数据结构选择：根据场景选择最合适的数据结构
4. 锁的演进：从synchronized -> Redis锁 -> Redisson锁
5. 拦截器分层：职责分离，代码清晰

可改进之处

• 可以引入本地缓存（Caffeine）进一步减轻Redis压力
• Feed流可以考虑推拉结合模式
• 可以增加Sentinel或Hystrix进行熔断降级
• 考虑引入Canal监听binlog，实现缓存与数据库的最终一致性

---

项目源码：heimadianping