编程导航会员会员
登录/注册

亿级流量点赞系统-第五章

1. 上期回顾

项目源码地址：https://github.com/ruogu-coder/ruogu-like

在第四章，使用 HeavyKeeper 算法检测热点 Key，并且使用本地缓存 Caffeine 存储热点 key, 减轻 Redis 的压力。

2. 本期目标

引入消息队列优化系统架构，提升削峰填谷的能力和系统可用性

3. 开发实现

3.1. 安装 Pulsar

在自己的虚拟机或者云服务器上安装 docker 然后执行如下命令：

▼
java
复制代码
docker run --name pulsar -p6650:6650 -p8080:8080 --mount source=pulsardata,target=/pulsar/data --mount source=pulsarconf,target=/pulsar/conf -d apachepulsar/pulsar:4.0.3 bin/pulsar standalone

3.2. 引入依赖

在 pom.xml 中引入如下依赖：

▼
xml
复制代码
<!-- 引入 spring-pulsar -->  
<dependency>  
  <groupId>org.springframework.boot</groupId>  
  <artifactId>spring-boot-starter-pulsar</artifactId>  
</dependency>

3.3. 配置信息

根据自己的虚拟机 IP 补充 pulsar 的 ip + 端口号

▼
java
复制代码
spring:
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://localhost:3306/ruogu_like?useSSL=false&allowPublicKeyRetrieval=true&useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai
    username: root
    password: 123456
  data:
    redis:
      host: localhost
      port: 6379
      database: 0
      connect-timeout: 5000
  pulsar:
    client:
      service-url: pulsar://192.168.44.128:6650
    admin:
      service-url: http://192.168.44.128:8080

3.4. 代码实现

3.4.1. 创建事件对象

创建listener.thumb.msg包，并创建ThumbEvent

▼
java
复制代码
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ThumbEvent implements Serializable {

    /**
     * 用户ID
     */
    private Long userId;

    /**
     * 博客ID
     */
    private Long blogId;

    /**
     * 事件类型
     */
    private EventType type;

    /**
     * 事件发生时间
     */
    private LocalDateTime eventTime;

    /**
     * 事件类型枚举
     */
    public enum EventType {
        /**
         * 点赞
         */
        INCR,

        /**
         * 取消点赞
         */
        DECR
    }
}

3.4.2. 新增 Lua 脚本

在constant.RedisLuaScriptConstant新增脚本定义：

▼
java
复制代码
/**
     * 点赞 Lua 脚本
     * KEYS[1]       -- 用户点赞状态键
     * ARGV[1]       -- 博客 ID
     * 返回:
     * -1: 已点赞
     * 1: 操作成功
     */
    public static final RedisScript<Long> THUMB_SCRIPT_MQ = new DefaultRedisScript<>("""  
                    local userThumbKey = KEYS[1]
                    local blogId = ARGV[1]
            
                    -- 判断是否已经点赞
                    if redis.call("HEXISTS", userThumbKey, blogId) == 1 then
                        return -1
                    end
            
                    -- 添加点赞记录
                    redis.call("HSET", userThumbKey, blogId, 1)
                    return 1
            """, Long.class);

    /**
     * 取消点赞 Lua 脚本
     * KEYS[1]       -- 用户点赞状态键
     * ARGV[1]       -- 博客 ID
     * 返回:
     * -1: 已点赞
     * 1: 操作成功
     */
    public static final RedisScript<Long> UN_THUMB_SCRIPT_MQ = new DefaultRedisScript<>("""  
            local userThumbKey = KEYS[1]
            local blogId = ARGV[1]
            
            -- 判断是否已点赞
            if redis.call("HEXISTS", userThumbKey, blogId) == 0 then
                return -1
            end
            
            -- 删除点赞记录
            redis.call("HDEL", userThumbKey, blogId)
            return 1
            """, Long.class);

3.4.3. 消息队列配置类

在config包下新建ThumbConsumerConfig 配置消费者批量处理策略：

▼
java
复制代码
@Configuration
public class ThumbConsumerConfig<T> implements PulsarListenerConsumerBuilderCustomizer<T> {
    @Override
    public void customize(ConsumerBuilder<T> consumerBuilder) {
        consumerBuilder.batchReceivePolicy(
                BatchReceivePolicy.builder()
                        // 每次处理 1000 条
                        .maxNumMessages(1000)
                        // 设置超时时间（单位：毫秒）
                        .timeout(10000, TimeUnit.MILLISECONDS)
                        .build()
        );
    }
}

3.4.4. 服务创建

将原来的ThumbLocalServiceImpl修改@Service("thumbServiceLocalCache")，创建ThumbServiceMQImpl

▼
java
复制代码
@Service("thumbService")
@Slf4j
@RequiredArgsConstructor
public class ThumbServiceMQImpl extends ServiceImpl<ThumbMapper, Thumb> implements ThumbService {
    
    private final UserService userService;

    private final RedisTemplate<String, Object> redisTemplate;

    private final PulsarTemplate<ThumbEvent> pulsarTemplate;

    @Override
    public Boolean doThumb(ThumbLikeOrUnLikeDTO thumbLikeOrUnLikeDTO) {
        if (thumbLikeOrUnLikeDTO == null || thumbLikeOrUnLikeDTO.getBlogId() == null) {
            throw exception(BAD_REQUEST);
        }
        User loginUser = userService.getLoginUser();
        if (loginUser == null) {
            throw exception(UNAUTHORIZED);
        }
        Long loginUserId = loginUser.getId();
        Long blogId = thumbLikeOrUnLikeDTO.getBlogId();
        String userThumbKey = RedisKeyUtil.getUserThumbKey(loginUserId);
        // 执行 Lua 脚本，点赞存入 Redis
        long result = redisTemplate.execute(
                RedisLuaScriptConstant.THUMB_SCRIPT_MQ,
                List.of(userThumbKey),
                blogId
        );
        if (LuaStatusEnum.FAIL.getValue() == result) {
            throw exception(USER_LIKE_ERROR);
        }

        ThumbEvent thumbEvent = ThumbEvent.builder()
                .blogId(blogId)
                .userId(loginUserId)
                .type(ThumbEvent.EventType.INCR)
                .eventTime(LocalDateTime.now())
                .build();
        pulsarTemplate.sendAsync("thumb-topic", thumbEvent).exceptionally(ex -> {
            redisTemplate.opsForHash().delete(userThumbKey, blogId.toString(), true);
            log.error("点赞事件发送失败: userId={}, blogId={}", loginUserId, blogId, ex);
            return null;
        });

        return true;
    }


    @Override
    public Boolean undoThumb(ThumbLikeOrUnLikeDTO thumbLikeOrUnLikeDTO) {
        if (thumbLikeOrUnLikeDTO == null || thumbLikeOrUnLikeDTO.getBlogId() == null) {
            throw exception(BAD_REQUEST);
        }
        User loginUser = userService.getLoginUser();
        if (loginUser == null) {
            throw exception(UNAUTHORIZED);
        }
        Long loginUserId = loginUser.getId();
        Long blogId = thumbLikeOrUnLikeDTO.getBlogId();
        String userThumbKey = RedisKeyUtil.getUserThumbKey(loginUserId);
        // 执行 Lua 脚本，点赞记录从 Redis 删除
        long result = redisTemplate.execute(
                RedisLuaScriptConstant.UN_THUMB_SCRIPT_MQ,
                List.of(userThumbKey),
                blogId
        );
        if (LuaStatusEnum.FAIL.getValue() == result) {
            throw exception(USER_UNLIKE_ERROR);
        }
        ThumbEvent thumbEvent = ThumbEvent.builder()
                .blogId(blogId)
                .userId(loginUserId)
                .type(ThumbEvent.EventType.DECR)
                .eventTime(LocalDateTime.now())
                .build();
        pulsarTemplate.sendAsync("thumb-topic", thumbEvent).exceptionally(ex -> {
            redisTemplate.opsForHash().put(userThumbKey, blogId.toString(), true);
            log.error("点赞事件发送失败: userId={}, blogId={}", loginUserId, blogId, ex);
            return null;
        });

        return true;
    }

    @Override
    public Boolean isThumb(Long blogId, Long userId) {
        return redisTemplate.opsForHash().hasKey(RedisKeyUtil.getUserThumbKey(userId), blogId.toString());
    }

}

目前这样修改后，每次点赞通过 Lua 脚本修改 Redis 的点赞信息，然后将点赞的持久化功能发送给了消息队列，让消息队列去消费，所以还需要实现消息队列的消费者。

3.4.5. 消息消费者

在listener.thumb创建ThumbConsumer

▼
java
复制代码
@Service
@RequiredArgsConstructor
@Slf4j
public class ThumbConsumer {

    private final BlogMapper blogMapper;
    private final ThumbService thumbService;

    // 批量处理配置
    @PulsarListener(
            subscriptionName = "thumb-subscription",
            topics = "thumb-topic",
            schemaType = SchemaType.JSON,
            batch = true,
            subscriptionType = SubscriptionType.Shared
            consumerCustomizer = "thumbConsumerConfig"
    )
    @Transactional(rollbackFor = Exception.class)
    public void processBatch(List<Message<ThumbEvent>> messages) {
        log.info("ThumbConsumer processBatch: {}", messages.size());
        Map<Long, Long> countMap = new ConcurrentHashMap<>();
        List<Thumb> thumbs = new ArrayList<>();

        // 并行处理消息
        LambdaQueryWrapper<Thumb> wrapper = new LambdaQueryWrapper<>();
        AtomicReference<Boolean> needRemove = new AtomicReference<>(false);

        // 提取事件并过滤无效消息
        List<ThumbEvent> events = messages.stream()
                .map(Message::getValue)
                .filter(Objects::nonNull)
                .toList();


        // 按(userId, blogId)分组，并获取每个分组的最新事件
        Map<Pair<Long, Long>, ThumbEvent> latestEvents = events.stream()
                .collect(Collectors.groupingBy(
                        e -> Pair.of(e.getUserId(), e.getBlogId()),
                        Collectors.collectingAndThen(
                                Collectors.toList(),
                                list -> {
                                    // 按时间升序排序，取最后一个作为最新事件
                                    list.sort(Comparator.comparing(ThumbEvent::getEventTime));
                                    if (list.size() % 2 == 0) {
                                        return null;
                                    }
                                    return list.getLast();
                                }
                        )
                ));

        latestEvents.forEach((userBlogPair, event) -> {
            if (event == null) {
                return;
            }
            ThumbEvent.EventType finalAction = event.getType();

            if (finalAction == ThumbEvent.EventType.INCR) {
                countMap.merge(event.getBlogId(), 1L, Long::sum);
                Thumb thumb = new Thumb();
                thumb.setBlogId(event.getBlogId());
                thumb.setUserId(event.getUserId());
                thumbs.add(thumb);
            } else {
                needRemove.set(true);
                wrapper.or().eq(Thumb::getUserId, event.getUserId()).eq(Thumb::getBlogId, event.getBlogId());
                countMap.merge(event.getBlogId(), -1L, Long::sum);
            }
        });

        // 批量更新数据库
        if (needRemove.get()) {
            thumbService.remove(wrapper);
        }
        batchUpdateBlogs(countMap);
        batchInsertThumbs(thumbs);
    }

    public void batchUpdateBlogs(Map<Long, Long> countMap) {
        if (!countMap.isEmpty()) {
            blogMapper.batchUpdateThumbCount(countMap);
        }
    }

    public void batchInsertThumbs(List<Thumb> thumbs) {
        if (!thumbs.isEmpty()) {
            // 分批次插入
            thumbService.saveBatch(thumbs, 500);
        }
    }
}

注意，需要在@PulsarListener注解的consumerCustomizer属性中，声明自定义的配置类thumbConsumerConfig这样批量处理的配置才能生效。

3.4.6. 测试

启动项目进行测试，能否正常启动。

  1. 点赞测试

  1. 取消点赞

3.4.7. 消息的可靠性保证

消息重试和死信队列

▼
java
复制代码
@Configuration
public class ThumbConsumerConfig<T> implements PulsarListenerConsumerBuilderCustomizer<T> {
    @Override
    public void customize(ConsumerBuilder<T> consumerBuilder) {
        consumerBuilder.batchReceivePolicy(
                BatchReceivePolicy.builder()
                        // 每次处理 1000 条
                        .maxNumMessages(1000)
                        // 设置超时时间（单位：毫秒）
                        .timeout(10000, TimeUnit.MILLISECONDS)
                        .build()
        );
    }

    // 配置 NACK 重试策略
    @Bean
    public RedeliveryBackoff negativeAckRedeliveryBackoff() {
        return MultiplierRedeliveryBackoff.builder()
                // 初始延迟 1 秒
                .minDelayMs(1000)
                // 最大延迟 60 秒
                .maxDelayMs(60_000)
                // 每次重试延迟倍数
                .multiplier(2)
                .build();
    }

    // 配置 ACK 超时重试策略
    @Bean
    public RedeliveryBackoff ackTimeoutRedeliveryBackoff() {
        return MultiplierRedeliveryBackoff.builder()
                // 初始延迟 5 秒
                .minDelayMs(5000)
                // 最大延迟 300 秒
                .maxDelayMs(300_000)
                .multiplier(3)
                .build();
    }

    @Bean
    public DeadLetterPolicy deadLetterPolicy() {
        return DeadLetterPolicy.builder()
                // 最大重试次数
                .maxRedeliverCount(3)
                // 死信主题名称
                .deadLetterTopic("thumb-dlq-topic")
                .build();
    }


}

在 ThumbConsumer 中，添加死信队列的处理方法

▼
java
复制代码
@PulsarListener(topics = "thumb-dlq-topic")
    public void consumerDlq(Message<ThumbEvent> message) {
        MessageId messageId = message.getMessageId();
        log.info("dlq message = {}", messageId);
        log.info("消息 {} 已入库", messageId);
        log.info("已通知相关人员 {} 处理消息 {}", "坤哥", messageId);
    }

批量处理配置方法

▼
java
复制代码
@PulsarListener(
            subscriptionName = "thumb-subscription",
            topics = "thumb-topic",
            schemaType = SchemaType.JSON,
            batch = true,
            subscriptionType = SubscriptionType.Shared,
            consumerCustomizer = "thumbConsumerConfig",
            // 引用 NACK 重试策略
            negativeAckRedeliveryBackoff = "negativeAckRedeliveryBackoff",
            // 引用 ACK 超时重试策略
            ackTimeoutRedeliveryBackoff = "ackTimeoutRedeliveryBackoff",
            // 引用死信队列策略
            deadLetterPolicy = "deadLetterPolicy"
    )
    @Transactional(rollbackFor = Exception.class)
    public void processBatch(List<Message<ThumbEvent>> messages) {
        log.info("ThumbConsumer processBatch: {}", messages.size());
        Map<Long, Long> countMap = new ConcurrentHashMap<>();
        List<Thumb> thumbs = new ArrayList<>();

        // 并行处理消息
        LambdaQueryWrapper<Thumb> wrapper = new LambdaQueryWrapper<>();
        AtomicReference<Boolean> needRemove = new AtomicReference<>(false);

        // 提取事件并过滤无效消息
        List<ThumbEvent> events = messages.stream()
                .map(Message::getValue)
                .filter(Objects::nonNull)
                .toList();


        // 按(userId, blogId)分组，并获取每个分组的最新事件
        Map<Pair<Long, Long>, ThumbEvent> latestEvents = events.stream()
                .collect(Collectors.groupingBy(
                        e -> Pair.of(e.getUserId(), e.getBlogId()),
                        Collectors.collectingAndThen(
                                Collectors.toList(),
                                list -> {
                                    // 按时间升序排序，取最后一个作为最新事件
                                    list.sort(Comparator.comparing(ThumbEvent::getEventTime));
                                    if (list.size() % 2 == 0) {
                                        return null;
                                    }
                                    return list.getLast();
                                }
                        )
                ));

        latestEvents.forEach((userBlogPair, event) -> {
            if (event == null) {
                return;
            }
            ThumbEvent.EventType finalAction = event.getType();

            if (finalAction == ThumbEvent.EventType.INCR) {
                countMap.merge(event.getBlogId(), 1L, Long::sum);
                Thumb thumb = new Thumb();
                thumb.setBlogId(event.getBlogId());
                thumb.setUserId(event.getUserId());
                thumbs.add(thumb);
            } else {
                needRemove.set(true);
                wrapper.or().eq(Thumb::getUserId, event.getUserId()).eq(Thumb::getBlogId, event.getBlogId());
                countMap.merge(event.getBlogId(), -1L, Long::sum);
            }
        });

        // 批量更新数据库
        if (needRemove.get()) {
            thumbService.remove(wrapper);
        }
        batchUpdateBlogs(countMap);
        batchInsertThumbs(thumbs);
    }

数据一致性保证

对账任务的实现，在 job 包下创建对账定时任务

▼
java
复制代码
@Service
@Slf4j
public class ThumbReconcileJob {
    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private ThumbService thumbService;

    @Resource
    private PulsarTemplate<ThumbEvent> pulsarTemplate;

    /**
     * 定时任务入口（每天凌晨2点执行）  
     */
    @Scheduled(cron = "0 0 2 * * ?")
    public void run() {
        long startTime = System.currentTimeMillis();

        // 1. 获取该分片下的所有用户ID  
        Set<Long> userIds = new HashSet<>();
        String pattern = ThumbConstant.USER_THUMB_KEY_PREFIX + "*";
        try (Cursor<String> cursor = redisTemplate.scan(ScanOptions.scanOptions().match(pattern).count(1000).build())) {
            while (cursor.hasNext()) {
                String key = cursor.next();
                Long userId = Long.valueOf(key.replace(ThumbConstant.USER_THUMB_KEY_PREFIX, ""));
                userIds.add(userId);
            }
        }

        // 2. 逐用户比对  
        userIds.forEach(userId -> {
            Set<Long> redisBlogIds = redisTemplate.opsForHash().keys(ThumbConstant.USER_THUMB_KEY_PREFIX + userId).stream().map(obj -> Long.valueOf(obj.toString())).collect(Collectors.toSet());
            Set<Long> mysqlBlogIds = Optional.ofNullable(thumbService.lambdaQuery()
                            .eq(Thumb::getUserId, userId)
                            .list()
                    ).orElse(new ArrayList<>())
                    .stream()
                    .map(Thumb::getBlogId)
                    .collect(Collectors.toSet());

            // 3. 计算差异（Redis有但MySQL无）  
            Set<Long> diffBlogIds = Sets.difference(redisBlogIds, mysqlBlogIds);

            // 4. 发送补偿事件  
            sendCompensationEvents(userId, diffBlogIds);
        });

        log.info("对账任务完成，耗时 {}ms", System.currentTimeMillis() - startTime);
    }

    /**
     * 发送补偿事件到Pulsar  
     */
    private void sendCompensationEvents(Long userId, Set<Long> blogIds) {
        blogIds.forEach(blogId -> {
            ThumbEvent thumbEvent = new ThumbEvent(userId, blogId, ThumbEvent.EventType.INCR, LocalDateTime.now());
            pulsarTemplate.sendAsync("thumb-topic", thumbEvent)
                    .exceptionally(ex -> {
                        log.error("补偿事件发送失败: userId={}, blogId={}", userId, blogId, ex);
                        return null;
                    });
        });
    }
}

4. 测试

  1. 消息处理异常重试

为了演示消息处理异常，这里在消费逻辑中抛出异常

▼
java
复制代码
public void processBatch(List<Message<ThumbEvent>> messages) {  
    log.info("ThumbConsumer processBatch: {}", messages.size());  
    for (Message<ThumbEvent> message : messages) {  
        log.info("message.getMessageId() = {}", message.getMessageId());  
    }  
    if (true) {  
        throw new RuntimeException("ThumbConsumer processBatch failed");  
    }  
}

通过日志我们可以发现一个问题，重试的间隔固定在十秒，并不是我们配置的 1 秒后重试

主要和这个配置有关系，这个配置让消费者每次处理 1000 条数据，当调用 Consumer#batchReceive() 方法时，消费者会根据 BatchReceivePolicy 的配置来决定如何批量接收消息：

如果消息队列中有足够的消息满足 maxNumMessages消费者会立即返回这些消息。

如果消息队列中的消息数量不足，消费者会根据 timeout 参数决定是否等待以及等待多长时间。如果 timeout 为0，消费者会立即返回，即使消息数量不足。如果 timeout 大于0，消费者会等待一段时间，直到有足够的消息或者超时时间到达

▼
java
复制代码
public void customize(ConsumerBuilder<T> consumerBuilder) {
        consumerBuilder.batchReceivePolicy(
                BatchReceivePolicy.builder()
                        // 每次处理 1000 条
                        .maxNumMessages(1000)
                        // 设置超时时间（单位：毫秒）
                        .timeout(10000, TimeUnit.MILLISECONDS)
                        .build()
        );
    }

配置分析

在我的配置中：

  • maxNumMessages(1000)：表示每次批量接收消息时最多接收1000条消息。
  • timeout(10000, TimeUnit.MILLISECONDS)：表示每次批量接收消息时的超时时间为10秒。

根据这些配置，当调用 Consumer#batchReceive() 方法时，消费者会尝试批量接收最多1000条消息。如果消息队列中的消息数量不足1000条，消费者会等待最多10秒，直到有足够的消息或者超时时间到达。

亿级流量点赞系统
0个评论
全部评论
点击登录，快来和大家讨论吧～
表情
图片
暂无评论
若谷
若谷
等级vip
内容推荐
更多
亿级流量点赞系统第九章
3
亿级流量点赞系统第十章
3
亿级流量点赞系统终章
8
#提问 亿级流量点赞项目亿级流量点赞项目，为啥项目拉下来就启动不了，好奇怪啊，这个是lombok报错，但是这些依赖正常都是有的，好奇怪啊，有人遇到同样的问题了吗？
1
📚 学习实战 | 亿级流量高并发点赞系统（Spring Boot3 + Java21 进阶项目落地）
8
作者分享
更多
AI超级智能体 - 一、接入AI大模型
9
亿级流量点赞系统-最终篇
5
亿级流量点赞系统-第十章
4
亿级流量点赞系统-第九章
6
亿级流量点赞系统-第八章
5
下载 APP