feat(count): 引入批量消费机制优化粉丝计数处理

- 添加 BufferTrigger依赖以支持消息聚合
- 实现消息批量消费逻辑,提升处理效率
- 配置批量大小为 1000,缓存队列最大容量为50000
- 设置聚合间隔为1 秒,均衡实时性与性能
- 新增测试用例验证大量消息发送与消费流程
- 日志记录聚合消息内容,便于调试与监控
This commit is contained in:
2025-10-15 22:33:20 +08:00
parent 31ab7c3d86
commit c6ac7193c1
4 changed files with 67 additions and 1 deletions

View File

@@ -97,6 +97,13 @@
<artifactId>rocketmq-spring-boot-starter</artifactId>
</dependency>
<!-- 快手 Buffer Trigger -->
<dependency>
<groupId>com.github.phantomthief</groupId>
<artifactId>buffer-trigger</artifactId>
</dependency>
</dependencies>
<build>

View File

@@ -1,11 +1,16 @@
package com.hanserwei.hannote.count.biz.consumer;
import com.github.phantomthief.collection.BufferTrigger;
import com.hanserwei.framework.common.utils.JsonUtils;
import com.hanserwei.hannote.count.biz.constant.MQConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.List;
@Component
@RocketMQMessageListener(
consumerGroup = "han_note_group_" + MQConstants.TOPIC_COUNT_FANS,
@@ -13,8 +18,20 @@ import org.springframework.stereotype.Component;
)
@Slf4j
public class CountFansConsumer implements RocketMQListener<String> {
private final BufferTrigger<String> bufferTrigger = BufferTrigger.<String>batchBlocking()
.bufferSize(50000) // 缓存队列的最大容量
.batchSize(1000) // 一批次最多聚合 1000 条
.linger(Duration.ofSeconds(1)) // 多久聚合一次
.setConsumerEx(this::consumeMessage)
.build();
@Override
public void onMessage(String body) {
log.info("## 消费了 MQ [计数:粉丝数]: {}", body);
// 往 bufferTrigger 中添加元素
bufferTrigger.enqueue(body);
}
private void consumeMessage(List<String> bodys) {
log.info("==> 聚合消息, size: {}", bodys.size());
log.info("==> 聚合消息, {}", JsonUtils.toJsonString(bodys));
}
}