feat(data-align): 实现用户关注、粉丝及笔记发布数的数据对齐功能

- 新增 LUA 脚本实现布隆过滤器校验日增量数据
- 修改表结构将 t_data_align_note_publish_count_temp 的 note_id 替换为 user_id
-为 CreateTableXxlJob 添加事务管理确保表创建一致性
- 新增 FollowUnfollowMqDTO 和 NoteOperateMqDTO 用于消息传递
- 扩展 InsertRecordMapper 支持插入关注、粉丝和笔记发布计数记录
- 在 RedisKeyConstants 中新增多个布隆过滤器相关常量和构建方法
- 新增两个 RocketMQ 消费者处理用户关注/取关和笔记发布/删除事件
- 更新 HTTP 测试文件中的请求参数以适配最新接口逻辑

han-note-data-align/src/main/java/com/hanserwei/hannote/data/align/constant/MQConstants.java

@@ -12,4 +12,14 @@ public interface MQConstants {
      */
     String TOPIC_COUNT_NOTE_COLLECT = "CountNoteCollectTopic";
 
+    /**
+     * Topic: 笔记操作（发布、删除）
+     */
+    String TOPIC_NOTE_OPERATE = "NoteOperateTopic";
+
+    /**
+     * Topic: 关注数计数
+     */
+    String TOPIC_COUNT_FOLLOWING = "CountFollowingTopic";
+
 }

han-note-data-align/src/main/java/com/hanserwei/hannote/data/align/constant/RedisKeyConstants.java

@@ -12,6 +12,21 @@ public class RedisKeyConstants {
      */
     public static final String BLOOM_TODAY_NOTE_COLLECT_LIST_KEY = "bloom:dataAlign:note:collects:";
 
+    /**
+     * 布隆过滤器：日增量变更数据，用户笔记发布，删除 前缀
+     */
+    public static final String BLOOM_TODAY_USER_NOTE_OPERATOR_LIST_KEY = "bloom:dataAlign:user:note:operators:";
+
+    /**
+     * 布隆过滤器：日增量变更数据，用户关注数 前缀
+     */
+    public static final String BLOOM_TODAY_USER_FOLLOW_LIST_KEY = "bloom:dataAlign:user:follows:";
+
+    /**
+     * 布隆过滤器：日增量变更数据，用户粉丝数 前缀
+     */
+    public static final String BLOOM_TODAY_USER_FANS_LIST_KEY = "bloom:dataAlign:user:fans:";
+
 
     /**
      * 构建完整的布隆过滤器：日增量变更数据，用户笔记点赞，取消点赞 KEY
@@ -33,4 +48,34 @@ public class RedisKeyConstants {
         return BLOOM_TODAY_NOTE_COLLECT_LIST_KEY + date;
     }
 
+    /**
+     * 构建完整的布隆过滤器：日增量变更数据，用户笔记发布，删除 KEY
+     *
+     * @param date 日期
+     * @return 完整的布隆过滤器：日增量变更数据，用户笔记发布，删除 KEY
+     */
+    public static String buildBloomUserNoteOperateListKey(String date) {
+        return BLOOM_TODAY_USER_NOTE_OPERATOR_LIST_KEY + date;
+    }
+
+    /**
+     * 构建完整的布隆过滤器：日增量变更数据，用户关注数 KEY
+     *
+     * @param date 日期
+     * @return 完整的布隆过滤器：日增量变更数据，用户关注数 KEY
+     */
+    public static String buildBloomUserFollowListKey(String date) {
+        return BLOOM_TODAY_USER_FOLLOW_LIST_KEY + date;
+    }
+
+    /**
+     * 构建完整的布隆过滤器：日增量变更数据，用户粉丝数 KEY
+     *
+     * @param date 日期
+     * @return 完整的布隆过滤器：日增量变更数据，用户粉丝数 KEY
+     */
+    public static String buildBloomUserFansListKey(String date) {
+        return BLOOM_TODAY_USER_FANS_LIST_KEY + date;
+    }
+
 }

han-note-data-align/src/main/java/com/hanserwei/hannote/data/align/consumer/TodayNotePublishIncrementData2DBConsumer.java

@@ -0,0 +1,91 @@
+package com.hanserwei.hannote.data.align.consumer;
+
+import com.hanserwei.framework.common.utils.JsonUtils;
+import com.hanserwei.hannote.data.align.constant.MQConstants;
+import com.hanserwei.hannote.data.align.constant.RedisKeyConstants;
+import com.hanserwei.hannote.data.align.constant.TableConstants;
+import com.hanserwei.hannote.data.align.domain.mapper.InsertRecordMapper;
+import com.hanserwei.hannote.data.align.model.vo.NoteOperateMqDTO;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.script.DefaultRedisScript;
+import org.springframework.data.redis.core.script.RedisScript;
+import org.springframework.scripting.support.ResourceScriptSource;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
+import java.util.Collections;
+import java.util.Objects;
+
+@Component
+@Slf4j
+@RocketMQMessageListener(
+        consumerGroup = "han_note_group_data_align_" + MQConstants.TOPIC_NOTE_OPERATE,
+        topic = MQConstants.TOPIC_NOTE_OPERATE
+)
+public class TodayNotePublishIncrementData2DBConsumer implements RocketMQListener<String> {
+
+    @Resource
+    private RedisTemplate<String, Object> redisTemplate;
+
+    @Resource
+    private InsertRecordMapper insertRecordMapper;
+
+    /**
+     * 表总分片数
+     */
+    @Value("${table.shards}")
+    private int tableShards;
+
+    @Override
+    public void onMessage(String body) {
+        log.info("## TodayNotePublishIncrementData2DBConsumer 消费到了 MQ: {}", body);
+
+        // 1. 布隆过滤器判断该日增量数据是否已经记录
+        // 消息字符串转换为DTO类
+        NoteOperateMqDTO noteOperateMqDTO = JsonUtils.parseObject(body, NoteOperateMqDTO.class);
+
+        if (Objects.isNull(noteOperateMqDTO)) {
+            return;
+        }
+
+        // 发布、被删除的作品的作者Id
+        Long noteCreatorId = noteOperateMqDTO.getCreatorId();
+
+        // 今日日期
+        String date = LocalDate.now()
+                .format(DateTimeFormatter.ofPattern("yyyyMMdd"));
+
+        String bloomKey = RedisKeyConstants.buildBloomUserNoteOperateListKey(date);
+
+        // 1. 布隆过滤器判断该日增量数据是否已经记录
+        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
+        // Lua 脚本路径
+        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/bloom_today_user_note_publish_check.lua")));
+        // 返回值类型
+        script.setResultType(Long.class);
+
+        // 执行 Lua 脚本，拿到返回结果
+        Long result = redisTemplate.execute(script, Collections.singletonList(bloomKey), noteCreatorId);
+
+        // 若布隆过滤器判断不存在（绝对正确）
+        if (Objects.equals(result, 0L)) {
+            // 根据分片总数，取模，分别获取对应的分片序号
+            long userIdHashKey = noteCreatorId % tableShards;
+
+            // 将日增量变更数据，写入日增量表中
+            // - t_data_align_note_publish_count_temp_日期_分片序号
+            insertRecordMapper.insert2DataAlignUserNotePublishCountTempTable(TableConstants.buildTableNameSuffix(date, userIdHashKey), noteCreatorId);
+
+            // 3. 数据库写入成功后，再添加布隆过滤器中
+            RedisScript<Long> bloomAddScript = RedisScript.of("return redis.call('BF.ADD', KEYS[1], ARGV[1])", Long.class);
+            redisTemplate.execute(bloomAddScript, Collections.singletonList(bloomKey), noteCreatorId);
+        }
+    }
+}

han-note-data-align/src/main/java/com/hanserwei/hannote/data/align/consumer/TodayUserFollowIncrementData2DBConsumer.java

@@ -0,0 +1,129 @@
+package com.hanserwei.hannote.data.align.consumer;
+
+import com.hanserwei.framework.common.utils.JsonUtils;
+import com.hanserwei.hannote.data.align.constant.MQConstants;
+import com.hanserwei.hannote.data.align.constant.RedisKeyConstants;
+import com.hanserwei.hannote.data.align.constant.TableConstants;
+import com.hanserwei.hannote.data.align.domain.mapper.InsertRecordMapper;
+import com.hanserwei.hannote.data.align.model.vo.FollowUnfollowMqDTO;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.script.DefaultRedisScript;
+import org.springframework.data.redis.core.script.RedisScript;
+import org.springframework.scripting.support.ResourceScriptSource;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
+import java.util.Collections;
+import java.util.Objects;
+
+@Slf4j
+@Component
+@RocketMQMessageListener(
+        consumerGroup = "han_note_group_data_align_" + MQConstants.TOPIC_COUNT_FOLLOWING,
+        topic = MQConstants.TOPIC_COUNT_FOLLOWING
+)
+public class TodayUserFollowIncrementData2DBConsumer implements RocketMQListener<String> {
+
+    @Resource
+    private RedisTemplate<String, Object> redisTemplate;
+
+    @Resource
+    private InsertRecordMapper insertRecordMapper;
+
+    /**
+     * 表总分片数
+     */
+    @Value("${table.shards}")
+    private int tableShards;
+
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    public void onMessage(String body) {
+        log.info("## TodayUserFollowIncrementData2DBConsumer 消费到了 MQ: {}", body);
+
+        // 字符串转换为DTO对象
+        FollowUnfollowMqDTO followUnfollowMqDTO = JsonUtils.parseObject(body, FollowUnfollowMqDTO.class);
+
+        if (Objects.isNull(followUnfollowMqDTO)) {
+            return;
+        }
+
+        // 关注/取关操作
+        // 源用户 ID
+        Long userId = followUnfollowMqDTO.getUserId();
+        // 目标用户 ID
+        Long targetUserId = followUnfollowMqDTO.getTargetUserId();
+
+        // 今日日期
+        String date = LocalDate.now()
+                .format(DateTimeFormatter.ofPattern("yyyyMMdd")); // 转字符串
+
+        // ------------------------- 源用户的关注数变更记录 -------------------------
+        // 布隆过滤器判断该日增量数据是否已经记录
+        String userBloomKey = RedisKeyConstants.buildBloomUserFollowListKey(date);
+
+        // 1. 布隆过滤器判断该日增量数据是否已经记录
+        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
+        // Lua 脚本路径
+        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/bloom_today_user_follow_check.lua")));
+        // 返回值类型
+        script.setResultType(Long.class);
+
+        // 执行 Lua 脚本，拿到返回结果
+        Long result = redisTemplate.execute(script, Collections.singletonList(userBloomKey), userId);
+
+        // Lua 脚本：添加到布隆过滤器
+        RedisScript<Long> bloomAddScript = RedisScript.of("return redis.call('BF.ADD', KEYS[1], ARGV[1])", Long.class);
+
+        // 若布隆过滤器判断不存在（绝对正确）
+        if (Objects.equals(result, 0L)) {
+            // 根据分片总数，取模，分别获取对应的分片序号
+            long userIdHashKey = userId % tableShards;
+
+            try {
+                // 将日增量变更数据，写入表 t_data_align_following_count_temp_日期_分片序号
+                insertRecordMapper.insert2DataAlignUserFollowingCountTempTable(
+                        TableConstants.buildTableNameSuffix(date, userIdHashKey), userId);
+            } catch (Exception e) {
+                log.error("", e);
+                throw new RuntimeException(e);
+            }
+
+            // 数据库写入成功后，再添加布隆过滤器中
+            redisTemplate.execute(bloomAddScript, Collections.singletonList(userBloomKey), userId);
+        }
+
+        // ------------------------- 目标用户的粉丝数变更记录 -------------------------
+        // 目标用户 ID 对应的 Bloom Key
+        String targetUserBloomKey = RedisKeyConstants.buildBloomUserFansListKey(date);
+        // 布隆过滤器判断该日增量数据是否已经记录
+        result = redisTemplate.execute(script, Collections.singletonList(targetUserBloomKey), targetUserId);
+
+        // 若布隆过滤器判断不存在（绝对正确）
+        if (Objects.equals(result, 0L)) {
+            // 若无，才会落库，减轻数据库压力
+            // 根据分片总数，取模，分别获取对应的分片序号
+            long targetUserIdHashKey = targetUserId % tableShards;
+
+            try {
+                // 将日增量变更数据，写入表 t_data_align_fans_count_temp_日期_分片序号
+                insertRecordMapper.insert2DataAlignUserFansCountTempTable(
+                        TableConstants.buildTableNameSuffix(date, targetUserIdHashKey), targetUserId);
+            } catch (Exception e) {
+                log.error("", e);
+                throw new RuntimeException(e);
+            }
+
+            // 数据库写入成功后，再添加布隆过滤器中
+            redisTemplate.execute(bloomAddScript, Collections.singletonList(targetUserBloomKey), targetUserId);
+        }
+    }
+}

han-note-data-align/src/main/java/com/hanserwei/hannote/data/align/domain/mapper/InsertRecordMapper.java

@@ -26,4 +26,19 @@ public interface InsertRecordMapper {
      * 用户获得的收藏数：计数变更
      */
     void insert2DataAlignUserCollectCountTempTable(@Param("tableNameSuffix") String tableNameSuffix, @Param("userId") Long userId);
+
+    /**
+     * 用户已发布笔记数：计数变更
+     */
+    void insert2DataAlignUserNotePublishCountTempTable(@Param("tableNameSuffix") String tableNameSuffix, @Param("userId") Long userId);
+
+    /**
+     * 用户关注数：计数变更
+     */
+    void insert2DataAlignUserFollowingCountTempTable(@Param("tableNameSuffix") String tableNameSuffix, @Param("userId") Long userId);
+
+    /**
+     * 用户粉丝数：计数变更
+     */
+    void insert2DataAlignUserFansCountTempTable(@Param("tableNameSuffix") String tableNameSuffix, @Param("userId") Long userId);
 }

han-note-data-align/src/main/java/com/hanserwei/hannote/data/align/job/CreateTableXxlJob.java

@@ -5,13 +5,16 @@ import com.hanserwei.hannote.data.align.domain.mapper.CreateTableMapper;
 import com.xxl.job.core.context.XxlJobHelper;
 import com.xxl.job.core.handler.annotation.XxlJob;
 import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.cloud.context.config.annotation.RefreshScope;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.support.TransactionTemplate;
 
 import java.time.LocalDate;
 import java.time.format.DateTimeFormatter;
 
+@Slf4j
 @Component
 @RefreshScope
 @SuppressWarnings("unused")
@@ -26,6 +29,9 @@ public class CreateTableXxlJob {
     @Value("${table.shards}")
     private int tableShards;
 
+    @Resource
+    private TransactionTemplate transactionTemplate;
+
     /**
      * 1、简单任务示例（Bean模式）
      */
@@ -41,15 +47,24 @@ public class CreateTableXxlJob {
                 // 表名后缀
                 String tableNameSuffix = TableConstants.buildTableNameSuffix(date, hashKey);
 
-                // 创建表
+                transactionTemplate.execute(status -> {
-                // 创建表
+                    try {
-                createTableMapper.createDataAlignFollowingCountTempTable(tableNameSuffix);
+                        // 创建表
-                createTableMapper.createDataAlignFansCountTempTable(tableNameSuffix);
+                        createTableMapper.createDataAlignFollowingCountTempTable(tableNameSuffix);
-                createTableMapper.createDataAlignNoteCollectCountTempTable(tableNameSuffix);
+                        createTableMapper.createDataAlignFansCountTempTable(tableNameSuffix);
-                createTableMapper.createDataAlignUserCollectCountTempTable(tableNameSuffix);
+                        createTableMapper.createDataAlignNoteCollectCountTempTable(tableNameSuffix);
-                createTableMapper.createDataAlignUserLikeCountTempTable(tableNameSuffix);
+                        createTableMapper.createDataAlignUserCollectCountTempTable(tableNameSuffix);
-                createTableMapper.createDataAlignNoteLikeCountTempTable(tableNameSuffix);
+                        createTableMapper.createDataAlignUserLikeCountTempTable(tableNameSuffix);
-                createTableMapper.createDataAlignNotePublishCountTempTable(tableNameSuffix);
+                        createTableMapper.createDataAlignNoteLikeCountTempTable(tableNameSuffix);
+                        createTableMapper.createDataAlignNotePublishCountTempTable(tableNameSuffix);
+                        return true;
+                    } catch (Exception e) {
+                        status.setRollbackOnly();
+                        log.error("创建表失败", e);
+                    }
+                    return false;
+                });
+
             }
         }
         XxlJobHelper.log("## 创建日增量数据表成功，表名后缀: {}...", date);

han-note-data-align/src/main/java/com/hanserwei/hannote/data/align/model/vo/FollowUnfollowMqDTO.java

@@ -0,0 +1,29 @@
+package com.hanserwei.hannote.data.align.model.vo;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@Builder
+public class FollowUnfollowMqDTO {
+
+    /**
+     * 原用户
+     */
+    private Long userId;
+
+    /**
+     * 目标用户
+     */
+    private Long targetUserId;
+
+    /**
+     * 1:关注 0:取关
+     */
+    private Integer type;
+
+}

han-note-data-align/src/main/java/com/hanserwei/hannote/data/align/model/vo/NoteOperateMqDTO.java

@@ -0,0 +1,29 @@
+package com.hanserwei.hannote.data.align.model.vo;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@Builder
+public class NoteOperateMqDTO {
+
+    /**
+     * 笔记发布者 ID
+     */
+    private Long creatorId;
+
+    /**
+     * 笔记 ID
+     */
+    private Long noteId;
+
+    /**
+     * 操作类型： 0 - 笔记删除； 1：笔记发布；
+     */
+    private Integer type;
+
+}

han-note-data-align/src/main/resources/lua/bloom_today_user_follow_check.lua

@@ -0,0 +1,16 @@
+-- LUA 脚本：日增量用户关注、取关变更数据布隆过滤器
+
+local key = KEYS[1] -- 操作的 Redis Key
+local userId = ARGV[1] -- Redis Value
+
+-- 使用 EXISTS 命令检查布隆过滤器是否存在
+local exists = redis.call('EXISTS', key)
+if exists == 0 then
+    -- 创建布隆过滤器
+    redis.call('BF.ADD', key, '')
+    -- 设置过期时间，一天后过期
+    redis.call("EXPIRE", key, 20 * 60 * 60)
+end
+
+-- 校验该变更数据是否已经存在(1 表示已存在，0 表示不存在)
+return redis.call('BF.EXISTS', key, userId)

han-note-data-align/src/main/resources/lua/bloom_today_user_note_publish_check.lua

@@ -0,0 +1,16 @@
+-- LUA 脚本：日增量笔记发布、删除变更数据布隆过滤器
+
+local key = KEYS[1] -- 操作的 Redis Key
+local userId = ARGV[1] -- Redis Value
+
+-- 使用 EXISTS 命令检查布隆过滤器是否存在
+local exists = redis.call('EXISTS', key)
+if exists == 0 then
+    -- 创建布隆过滤器
+    redis.call('BF.ADD', key, '')
+    -- 设置过期时间，一天后过期
+    redis.call("EXPIRE", key, 20 * 60 * 60)
+end
+
+-- 校验该变更数据是否已经存在(1 表示已存在，0 表示不存在)
+return redis.call('BF.EXISTS', key, userId)

han-note-data-align/src/main/resources/mapperxml/CreateTableMapper.xml

@@ -76,9 +76,9 @@
         CREATE TABLE IF NOT EXISTS `t_data_align_note_publish_count_temp_${tableNameSuffix}`
         (
             `id`      bigint unsigned NOT NULL AUTO_INCREMENT COMMENT '主键ID',
-            `note_id` bigint unsigned NOT NULL COMMENT '笔记ID',
+            `user_id` bigint unsigned NOT NULL COMMENT '用户ID',
             PRIMARY KEY (`id`) USING BTREE,
-            UNIQUE KEY `uk_note_id` (`note_id`)
+            UNIQUE KEY `uk_user_id` (`user_id`)
         ) ENGINE = InnoDB
           DEFAULT CHARSET = utf8mb4
           COLLATE = utf8mb4_unicode_ci COMMENT ='数据对齐日增量表：用户发布笔记数';

han-note-data-align/src/main/resources/mapperxml/InsertRecordMapper.xml

@@ -20,4 +20,19 @@
         insert into `t_data_align_user_collect_count_temp_${tableNameSuffix}` (user_id)
         values (#{userId})
     </insert>
+
+    <insert id="insert2DataAlignUserNotePublishCountTempTable" parameterType="map">
+        insert into `t_data_align_note_publish_count_temp_${tableNameSuffix}` (user_id)
+        values (#{userId})
+    </insert>
+
+    <insert id="insert2DataAlignUserFollowingCountTempTable" parameterType="map">
+        insert into `t_data_align_following_count_temp_${tableNameSuffix}` (user_id)
+        values (#{userId})
+    </insert>
+
+    <insert id="insert2DataAlignUserFansCountTempTable" parameterType="map">
+        insert into `t_data_align_fans_count_temp_${tableNameSuffix}` (user_id)
+        values (#{userId})
+    </insert>
 </mapper>

http-client/gateApi.http

@@ -77,8 +77,8 @@ Authorization: Bearer {{token}}
   "imgUris": [
     "https://cdn.pixabay.com/photo/2025/10/05/15/06/autumn-9875155_1280.jpg"
   ],
-  "title": "第三篇图文笔记",
+  "title": "测试数据对齐图文笔记5",
-  "content": "这个是第三篇图文笔记的测试",
+  "content": "测试数据对齐测试数据对齐测试数据对齐测试5",
   "topicId": 1
 }
 
@@ -126,7 +126,7 @@ Content-Type: application/json
 Authorization: Bearer {{token}}
 
 {
-  "id": 1979849112022941780
+  "id": 1981322504056078370
 }
 
 ### 关注自己
@@ -153,7 +153,7 @@ Content-Type: application/json
 Authorization: Bearer {{token}}
 
 {
-  "followUserId": {{otherUserId}}
+  "followUserId": 2100
 }
 
 ### 取消关注