From c6ac7193c1c712f232adb5ef622f11c09d28f6b7 Mon Sep 17 00:00:00 2001
From: Hanserwei <2628273921@qq.com>
Date: Wed, 15 Oct 2025 22:33:20 +0800
Subject: [PATCH] =?UTF-8?q?feat(count):=20=E5=BC=95=E5=85=A5=E6=89=B9?=
 =?UTF-8?q?=E9=87=8F=E6=B6=88=E8=B4=B9=E6=9C=BA=E5=88=B6=E4=BC=98=E5=8C=96?=
 =?UTF-8?q?=E7=B2=89=E4=B8=9D=E8=AE=A1=E6=95=B0=E5=A4=84=E7=90=86?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 添加 BufferTrigger依赖以支持消息聚合
- 实现消息批量消费逻辑，提升处理效率
- 配置批量大小为 1000，缓存队列最大容量为50000
- 设置聚合间隔为1 秒，均衡实时性与性能
- 新增测试用例验证大量消息发送与消费流程
- 日志记录聚合消息内容，便于调试与监控
---
 han-note-count/han-note-count-biz/pom.xml     |  7 ++++
 .../count/biz/consumer/CountFansConsumer.java | 19 +++++++++-
 .../hannote/user/relation/biz/MQTests.java    | 35 +++++++++++++++++++
 pom.xml                                       |  7 ++++
 4 files changed, 67 insertions(+), 1 deletion(-)

diff --git a/han-note-count/han-note-count-biz/pom.xml b/han-note-count/han-note-count-biz/pom.xml
index ecc3570..cf6a488 100644
--- a/han-note-count/han-note-count-biz/pom.xml
+++ b/han-note-count/han-note-count-biz/pom.xml
@@ -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>
diff --git a/han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/consumer/CountFansConsumer.java b/han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/consumer/CountFansConsumer.java
index 2e1e149..fbcbaa8 100644
--- a/han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/consumer/CountFansConsumer.java
+++ b/han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/consumer/CountFansConsumer.java
@@ -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));
     }
 }
diff --git a/han-note-user-relation/han-note-user-relation-biz/src/test/java/com/hanserwei/hannote/user/relation/biz/MQTests.java b/han-note-user-relation/han-note-user-relation-biz/src/test/java/com/hanserwei/hannote/user/relation/biz/MQTests.java
index 364414d..8d0e3e7 100644
--- a/han-note-user-relation/han-note-user-relation-biz/src/test/java/com/hanserwei/hannote/user/relation/biz/MQTests.java
+++ b/han-note-user-relation/han-note-user-relation-biz/src/test/java/com/hanserwei/hannote/user/relation/biz/MQTests.java
@@ -2,6 +2,8 @@ package com.hanserwei.hannote.user.relation.biz;
 
 import com.hanserwei.framework.common.utils.JsonUtils;
 import com.hanserwei.hannote.user.relation.biz.constant.MQConstants;
+import com.hanserwei.hannote.user.relation.biz.enums.FollowUnfollowTypeEnum;
+import com.hanserwei.hannote.user.relation.biz.model.dto.CountFollowUnfollowMqDTO;
 import com.hanserwei.hannote.user.relation.biz.model.dto.FollowUserMqDTO;
 import com.hanserwei.hannote.user.relation.biz.model.dto.UnfollowUserMqDTO;
 import jakarta.annotation.Resource;
@@ -125,4 +127,37 @@ class MQTests {
         }
     }
 
+    /**
+     * 测试：发送计数 MQ, 以统计粉丝数
+     */
+    @Test
+    void testSendCountFollowUnfollowMQ() {
+        // 循环发送 3200 条 MQ
+        for (long i = 0; i < 3200; i++) {
+            // 构建消息体 DTO
+            CountFollowUnfollowMqDTO countFollowUnfollowMqDTO = CountFollowUnfollowMqDTO.builder()
+                    .userId(i + 1) // 关注者用户 ID
+                    .targetUserId(27L) // 目标用户
+                    .type(FollowUnfollowTypeEnum.FOLLOW.getCode())
+                    .build();
+
+            // 构建消息对象，并将 DTO 转成 Json 字符串设置到消息体中
+            org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(countFollowUnfollowMqDTO))
+                    .build();
+
+            // 发送 MQ 通知计数服务：统计粉丝数
+            rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_FANS, message, new SendCallback() {
+                @Override
+                public void onSuccess(SendResult sendResult) {
+                    log.info("==> 【计数服务：粉丝数】MQ 发送成功，SendResult: {}", sendResult);
+                }
+
+                @Override
+                public void onException(Throwable throwable) {
+                    log.error("==> 【计数服务：粉丝数】MQ 发送异常: ", throwable);
+                }
+            });
+        }
+
+    }
 }
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index 47f3af7..98df385 100755
--- a/pom.xml
+++ b/pom.xml
@@ -64,6 +64,7 @@
         <zookeeper.version>3.9.4</zookeeper.version>
         <rocketmq-spring-boot.version>2.3.4</rocketmq-spring-boot.version>
         <rocketmq-client.version>5.3.2</rocketmq-client.version>
+        <buffertrigger.version>0.2.21</buffertrigger.version>
     </properties>
     <dependencyManagement>
         <dependencies>
@@ -285,6 +286,12 @@
                 <artifactId>rocketmq-acl</artifactId>
                 <version>${rocketmq-client.version}</version>
             </dependency>
+            <!-- 快手 Buffer Trigger -->
+            <dependency>
+                <groupId>com.github.phantomthief</groupId>
+                <artifactId>buffer-trigger</artifactId>
+                <version>${buffertrigger.version}</version>
+            </dependency>
 
         </dependencies>
     </dependencyManagement>