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

亿级流量点赞系统第五章

项目地址

GitHub地址💥：https://github.com/tenyon61/likeboom

后端初始化模板💥：https://github.com/tenyon61/springboot3-demo/tree/single

快速开始

✨方案选型

引用消息队列进行异步化改造，提升削峰填谷能力和系统可用性。我们现使用Apache Pulsar作为消息队列中间件，和其他消息队列相比，Apache Pulsar 是一款高性能、分布式的消息队列与流处理平台，其架构设计融合了发布订阅模型、分层存储、多租户支持等特性，适用于多种复杂场景。

💥适用场景

  • 大规模分布式系统与多租户架构
  • 跨地域部署与高可用性
  • 长期消息存储与离线分析
  • 实时流处理与事件驱动架构
  • 弹性扩展与流量波动应对
  • 混合云与边缘计算

💥方案比对

Pulsar 尤其在需要多租户、跨地域、弹性扩展、长期存储的场景中优势显著。如果业务面临高并发、全球化部署、混合云架构或需要整合实时与离线处理，Pulsar 是理想选择。若场景简单（如中小规模单集群、仅需基础队列功能），可优先考虑 Kafka、RabbitMQ 等轻量方案；若需求复杂且追求架构灵活性，Pulsar 的设计能提供更强的扩展性和长期维护价值。

✨Pulsar安装

💥单机部署

alive:true代表单机部署完成

2025-04-27-d4ba6d.webp

💥引入依赖

pom.xml

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

application.yml

▼
yaml
复制代码
spring:  
  pulsar:
    client:
      service-url: pulsar://${pulsar.host}:6650

💥点赞事件实体

ThumbEvent.java

▼
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
    }
}

💥Lua脚本改造

RedisConstant.java

▼
java
复制代码
/**
 * 点赞 Lua 脚本
 * KEYS[1]       -- 用户点赞状态键
 * ARGV[1]       -- 博客 ID
 * 返回:
 * -1: 已点赞
 * 1: 操作成功
 */
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: 操作成功
 */
RedisScript<Long> UNTHUMB_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);

💥消息队列配置

ThumbConsumerConfig.java

▼
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();
    }

}

💥服务改造

点赞和持久化分离

ThumbServiceMQImpl.java

▼
java
复制代码
@Override
public Boolean doThumb(DoThumbDTO doThumbDTO) {
    if (doThumbDTO == null || doThumbDTO.getBlogId() == null) {
        throw new RuntimeException("参数错误");
    }
    User loginUser = userService.getLoginUser();
    Long loginUserId = loginUser.getId();
    Long blogId = doThumbDTO.getBlogId();
    String userThumbKey = RedisUtils.getUserThumbKey(loginUserId);
    // 执行 Lua 脚本，点赞存入 Redis
    long result = redisTemplate.execute(
            RedisConstant.THUMB_SCRIPT_MQ,
            List.of(userThumbKey),
            blogId
    );
    if (LuaStatusEnum.FAIL.getValue() == result) {
        throw new BusinessException(ErrorCode.OPERATION_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(DoThumbDTO doThumbDTO) {
    if (doThumbDTO == null || doThumbDTO.getBlogId() == null) {
        throw new RuntimeException("参数错误");
    }
    User loginUser = userService.getLoginUser();
    Long loginUserId = loginUser.getId();
    Long blogId = doThumbDTO.getBlogId();
    String userThumbKey = RedisUtils.getUserThumbKey(loginUserId);
    // 执行 Lua 脚本，点赞记录从 Redis 删除
    long result = redisTemplate.execute(
            RedisConstant.UNTHUMB_SCRIPT_MQ,
            List.of(userThumbKey),
            blogId
    );
    if (LuaStatusEnum.FAIL.getValue() == result) {
        throw new BusinessException(ErrorCode.OPERATION_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;
}

💥服务消费者

ThumbConsumer.java

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

    private final BlogMapper blogMapper;
    private final ThumbService thumbService;

    @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);
    }

    // 批量处理配置
    @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());
        // for (Message<ThumbEvent> message : messages) {
        //     log.info("message.getMessageId() = {}", message.getMessageId());
        // }
        // if (true) {
        //     throw new RuntimeException("ThumbConsumer processBatch failed");
        // }
        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);
        }
    }
}

💥对账任务

ThumbReconcileJob.java对账job

▼
java
复制代码
 @Scheduled(cron = "0 0 2 * * ?")
    public void run() {
        long startTime = System.currentTimeMillis();

        // 1. 获取该分片下的所有用户ID
        Set<Long> userIds = new HashSet<>();
        String pattern = RedisConstant.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(RedisConstant.USER_THUMB_KEY_PREFIX, ""));
                userIds.add(userId);
            }
        }

        // 2. 逐用户比对
        userIds.forEach(userId -> {
            Set<Long> redisBlogIds = redisTemplate.opsForHash().keys(RedisConstant.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", DateUtil.current() - 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;
                    });
        });
    }
}

功能验证

  • 当执行点赞操作时，系统会生成一条 ThumbEvent 消息，并将这些消息发送到 thumb-topic 主题中。

  • ThumbConsumer 类中的 processBatch 方法会订阅 thumb-topic 主题，当有新消息到达时，会触发该方法执行批量处理逻辑。

  • 当消息进入死信队列 thumb-dlq-topic 后，consumerDlq 方法会被触发，记录相关日志信息，如消息 ID、入库信息，并通知。

2025-04-28-39bb1d.webp

总的来说，就是从点赞事件封装成消息发送到 thumb-topic 主题，到ThumbConsumer 类会对消息进行批量处理，到更新数据库中的点赞记录和博客点赞数，并处理出现的异常和死信队列消息。

本章总结

使用Pulsar 作为消息中间件，给点赞/取消点赞事件提供高性能、高可靠的异步通信能力，以及通过对账 Job 的最终一致性兜底，形成了完整的数据一致性解决方案。

项目踩坑

待补充

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