Compare commits
3 Commits
5c4d8862a2
...
d1f756d5c8
| Author | SHA1 | Date | |
|---|---|---|---|
| d1f756d5c8 | |||
| ac65664dfe | |||
| 17123657f4 |
3
.idea/data_source_mapping.xml
generated
3
.idea/data_source_mapping.xml
generated
@@ -3,7 +3,8 @@
|
||||
<component name="DataSourcePerFileMappings">
|
||||
<file url="file://$APPLICATION_CONFIG_DIR$/consoles/db/5b969fbe-0f66-42be-8d30-ff21036ab8a4/console.sql" value="5b969fbe-0f66-42be-8d30-ff21036ab8a4" />
|
||||
<file url="file://$APPLICATION_CONFIG_DIR$/consoles/db/f2474a4a-e4f1-4afa-bd43-7ae7738b47c5/console.sql" value="f2474a4a-e4f1-4afa-bd43-7ae7738b47c5" />
|
||||
<file url="file://$PROJECT_DIR$/han-note-data-align/src/main/resources/mapperxml/InsertRecordMapper.xml" value="f2474a4a-e4f1-4afa-bd43-7ae7738b47c5" />
|
||||
<file url="file://$PROJECT_DIR$/han-note-data-align/src/main/resources/mapperxml/InsertMapper.xml" value="f2474a4a-e4f1-4afa-bd43-7ae7738b47c5" />
|
||||
<file url="file://$PROJECT_DIR$/han-note-data-align/src/main/resources/mapperxml/SelectRecordMapper.xml" value="f2474a4a-e4f1-4afa-bd43-7ae7738b47c5" />
|
||||
<file url="file://$PROJECT_DIR$/sql/createData.sql" value="f2474a4a-e4f1-4afa-bd43-7ae7738b47c5" />
|
||||
<file url="file://$PROJECT_DIR$/sql/createTable.sql" value="f2474a4a-e4f1-4afa-bd43-7ae7738b47c5" />
|
||||
<file url="file://$PROJECT_DIR$/sql/leafcreatetable.sql" value="c4c1f1dc-816f-4113-88d6-9ebd7677af82" />
|
||||
|
||||
@@ -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";
|
||||
|
||||
}
|
||||
@@ -3,24 +3,78 @@ package com.hanserwei.hannote.data.align.constant;
|
||||
public class RedisKeyConstants {
|
||||
|
||||
/**
|
||||
* 布隆过滤器:日增量变更数据,用户笔记点赞,取消点赞 前缀
|
||||
* 布隆过滤器:日增量变更数据,用户笔记点赞,取消点赞(笔记ID) 前缀
|
||||
*/
|
||||
public static final String BLOOM_TODAY_NOTE_LIKE_LIST_KEY = "bloom:dataAlign:note:likes:";
|
||||
public static final String BLOOM_TODAY_NOTE_LIKE_NOTE_ID_LIST_KEY = "bloom:dataAlign:note:like:noteIds";
|
||||
|
||||
/**
|
||||
* 布隆过滤器:日增量变更数据,用户笔记收藏,取消收藏 前缀
|
||||
* 布隆过滤器:日增量变更数据,用户笔记点赞,取消点赞(笔记发布者ID) 前缀
|
||||
*/
|
||||
public static final String BLOOM_TODAY_NOTE_COLLECT_LIST_KEY = "bloom:dataAlign:note:collects:";
|
||||
public static final String BLOOM_TODAY_NOTE_LIKE_USER_ID_LIST_KEY = "bloom:dataAlign:note:like:userIds";
|
||||
|
||||
/**
|
||||
* 布隆过滤器:日增量变更数据,用户笔记收藏,取消收藏(笔记ID) 前缀
|
||||
*/
|
||||
public static final String BLOOM_TODAY_NOTE_COLLECT_NOTE_ID_LIST_KEY = "bloom:dataAlign:note:collect:noteIds";
|
||||
|
||||
/**
|
||||
* 布隆过滤器:日增量变更数据,用户笔记收藏,取消收藏(笔记发布者ID) 前缀
|
||||
*/
|
||||
public static final String BLOOM_TODAY_NOTE_COLLECT_USER_ID_LIST_KEY = "bloom:dataAlign:note:collect:userIds";
|
||||
|
||||
/**
|
||||
* 布隆过滤器:日增量变更数据,用户笔记发布,删除 前缀
|
||||
*/
|
||||
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:";
|
||||
|
||||
/**
|
||||
* Hash Field: 关注总数
|
||||
*/
|
||||
public static final String FIELD_FOLLOWING_TOTAL = "followingTotal";
|
||||
|
||||
/**
|
||||
* 用户维度计数 Key 前缀
|
||||
*/
|
||||
private static final String COUNT_USER_KEY_PREFIX = "count:user:";
|
||||
|
||||
/**
|
||||
* 构建用户维度计数 Key
|
||||
*
|
||||
* @param userId 用户 ID
|
||||
* @return 用户维度计数 Key
|
||||
*/
|
||||
public static String buildCountUserKey(Long userId) {
|
||||
return COUNT_USER_KEY_PREFIX + userId;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 构建完整的布隆过滤器:日增量变更数据,用户笔记点赞,取消点赞 KEY
|
||||
* 构建完整的布隆过滤器:日增量变更数据,用户笔记点赞,取消点赞(笔记ID) KEY
|
||||
* @param date 日期
|
||||
* @return 完整的布隆过滤器:日增量变更数据,用户笔记点赞,取消点赞(笔记ID) KEY
|
||||
*/
|
||||
public static String buildBloomUserNoteLikeNoteIdListKey(String date) {
|
||||
return BLOOM_TODAY_NOTE_LIKE_NOTE_ID_LIST_KEY + date;
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建完整的布隆过滤器:日增量变更数据,用户笔记点赞,取消点赞(笔记发布者ID) KEY
|
||||
*
|
||||
* @param date 日期
|
||||
* @return 完整的布隆过滤器:日增量变更数据,用户笔记点赞,取消点赞 KEY
|
||||
* @return 完整的布隆过滤器:日增量变更数据,用户笔记点赞,取消点赞(笔记发布者ID) KEY
|
||||
*/
|
||||
public static String buildBloomUserNoteLikeListKey(String date) {
|
||||
return BLOOM_TODAY_NOTE_LIKE_LIST_KEY + date;
|
||||
public static String buildBloomUserNoteLikeUserIdListKey(String date) {
|
||||
return BLOOM_TODAY_NOTE_LIKE_USER_ID_LIST_KEY + date;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -29,8 +83,48 @@ public class RedisKeyConstants {
|
||||
* @param date 日期
|
||||
* @return 完整的布隆过滤器:日增量变更数据,用户笔记收藏,取消收藏 KEY
|
||||
*/
|
||||
public static String buildBloomUserNoteCollectListKey(String date) {
|
||||
return BLOOM_TODAY_NOTE_COLLECT_LIST_KEY + date;
|
||||
public static String buildBloomUserNoteCollectNoteIdListKey(String date) {
|
||||
return BLOOM_TODAY_NOTE_COLLECT_NOTE_ID_LIST_KEY + date;
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建完整的布隆过滤器:日增量变更数据,用户笔记收藏,取消收藏 KEY
|
||||
*
|
||||
* @param date 日期
|
||||
* @return 完整的布隆过滤器:日增量变更数据,用户笔记收藏,取消收藏 KEY
|
||||
*/
|
||||
public static String buildBloomUserNoteCollectUserIdListKey(String date) {
|
||||
return BLOOM_TODAY_NOTE_COLLECT_USER_ID_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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -4,7 +4,7 @@ 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.domain.mapper.InsertMapper;
|
||||
import com.hanserwei.hannote.data.align.model.vo.CollectUnCollectNoteMqDTO;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
@@ -17,7 +17,6 @@ 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.support.TransactionTemplate;
|
||||
|
||||
import java.time.LocalDate;
|
||||
import java.time.format.DateTimeFormatter;
|
||||
@@ -35,9 +34,7 @@ public class TodayNoteCollectIncrementData2DBConsumer implements RocketMQListene
|
||||
@Resource
|
||||
private RedisTemplate<String, Object> redisTemplate;
|
||||
@Resource
|
||||
private TransactionTemplate transactionTemplate;
|
||||
@Resource
|
||||
private InsertRecordMapper insertRecordMapper;
|
||||
private InsertMapper insertMapper;
|
||||
|
||||
/**
|
||||
* 表总分片数
|
||||
@@ -64,7 +61,10 @@ public class TodayNoteCollectIncrementData2DBConsumer implements RocketMQListene
|
||||
String date = LocalDate.now()
|
||||
.format(DateTimeFormatter.ofPattern("yyyyMMdd"));
|
||||
|
||||
String bloomKey = RedisKeyConstants.buildBloomUserNoteCollectListKey(date);
|
||||
// ------------------------- 笔记的收藏数变更记录 -------------------------
|
||||
|
||||
// 笔记对应的 Bloom Key
|
||||
String noteBloomKey = RedisKeyConstants.buildBloomUserNoteCollectNoteIdListKey(date);
|
||||
|
||||
// 1. 布隆过滤器判断该日增量数据是否已经记录
|
||||
DefaultRedisScript<Long> script = new DefaultRedisScript<>();
|
||||
@@ -74,38 +74,46 @@ public class TodayNoteCollectIncrementData2DBConsumer implements RocketMQListene
|
||||
script.setResultType(Long.class);
|
||||
|
||||
// 执行 Lua 脚本,拿到返回结果
|
||||
Long result = redisTemplate.execute(script, Collections.singletonList(bloomKey), noteId);
|
||||
log.info("布隆过滤器判断结果:{}", result);
|
||||
Long result = redisTemplate.execute(script, Collections.singletonList(noteBloomKey), noteId);
|
||||
|
||||
// Lua 脚本:添加到布隆过滤器
|
||||
RedisScript<Long> bloomAddScript = RedisScript.of("return redis.call('BF.ADD', KEYS[1], ARGV[1])", Long.class);
|
||||
|
||||
// 若布隆过滤器判断不存在(绝对正确)
|
||||
if (Objects.equals(result, 0L)) {
|
||||
// 2. 若无,才会落库,减轻数据库压力
|
||||
// 根据分片总数,取模,分别获取对应的分片序号
|
||||
long userIdHashKey = noteCreatorId % tableShards;
|
||||
// 若无,才会落库数据库
|
||||
|
||||
// 根据分片总数,取模,获取对应的分片序号
|
||||
long noteIdHashKey = noteId % tableShards;
|
||||
log.info("根据分片总数,取模,分别获取对应的分片序号user:{},note:{}", userIdHashKey, noteIdHashKey);
|
||||
|
||||
// 编程式事务,保证多语句的原子性
|
||||
transactionTemplate.execute(status -> {
|
||||
try {
|
||||
// 将日增量变更数据,分别写入两张表
|
||||
// - t_data_align_note_collect_count_temp_日期_分片序号
|
||||
// - t_data_align_user_collect_count_temp_日期_分片序号
|
||||
insertRecordMapper.insert2DataAlignNoteCollectCountTempTable(TableConstants.buildTableNameSuffix(date, noteIdHashKey), noteId);
|
||||
insertRecordMapper.insert2DataAlignUserCollectCountTempTable(TableConstants.buildTableNameSuffix(date, userIdHashKey), noteCreatorId);
|
||||
insertMapper.insert2DataAlignNoteCollectCountTempTable(TableConstants.buildTableNameSuffix(date, noteIdHashKey), noteId);
|
||||
} catch (Exception e) {
|
||||
log.error("## TodayNoteCollectIncrementData2DBConsumer 笔记收藏数变更记录失败:{}", e.getMessage());
|
||||
}
|
||||
|
||||
return true;
|
||||
} catch (Exception ex) {
|
||||
status.setRollbackOnly(); // 标记事务为回滚
|
||||
log.error("", ex);
|
||||
// 数据库落库成功后,再添加布隆过滤器中
|
||||
redisTemplate.execute(bloomAddScript, Collections.singletonList(noteBloomKey), noteId);
|
||||
}
|
||||
return false;
|
||||
});
|
||||
|
||||
// 3. 数据库写入成功后,再添加布隆过滤器中
|
||||
// 4. 数据库写入成功后,再添加布隆过滤器中
|
||||
RedisScript<Long> bloomAddScript = RedisScript.of("return redis.call('BF.ADD', KEYS[1], ARGV[1])", Long.class);
|
||||
redisTemplate.execute(bloomAddScript, Collections.singletonList(bloomKey), noteId);
|
||||
// ------------------------- 笔记作者的收藏数变更记录 -------------------------
|
||||
// 笔记作者对应的 Bloom Key
|
||||
String userBloomKey = RedisKeyConstants.buildBloomUserNoteCollectUserIdListKey(date);
|
||||
// 执行 Lua 脚本,拿到返回结果
|
||||
result = redisTemplate.execute(script, Collections.singletonList(userBloomKey), noteCreatorId);
|
||||
// 若布隆过滤器判断不存在(绝对正确)
|
||||
if (Objects.equals(result, 0L)) {
|
||||
// 若无,才会落库数据库
|
||||
|
||||
// 根据分片总数,取模,获取对应的分片序号
|
||||
long noteCreatorIdHashKey = noteCreatorId % tableShards;
|
||||
try {
|
||||
insertMapper.insert2DataAlignUserCollectCountTempTable(TableConstants.buildTableNameSuffix(date, noteCreatorIdHashKey), noteCreatorId);
|
||||
} catch (Exception e) {
|
||||
log.error("## TodayNoteCollectIncrementData2DBConsumer 笔记作者的收藏数变更记录失败:{}", e.getMessage());
|
||||
}
|
||||
// 数据库落库成功后,再添加布隆过滤器中
|
||||
redisTemplate.execute(bloomAddScript, Collections.singletonList(userBloomKey), noteCreatorId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,7 @@ 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.domain.mapper.InsertMapper;
|
||||
import com.hanserwei.hannote.data.align.model.vo.LikeUnlikeNoteMqDTO;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
@@ -17,7 +17,6 @@ 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.support.TransactionTemplate;
|
||||
|
||||
import java.time.LocalDate;
|
||||
import java.time.format.DateTimeFormatter;
|
||||
@@ -35,35 +34,35 @@ public class TodayNoteLikeIncrementData2DBConsumer implements RocketMQListener<S
|
||||
@Resource
|
||||
private RedisTemplate<String, Object> redisTemplate;
|
||||
@Resource
|
||||
private TransactionTemplate transactionTemplate;
|
||||
@Resource
|
||||
private InsertRecordMapper insertRecordMapper;
|
||||
private InsertMapper insertMapper;
|
||||
|
||||
/**
|
||||
* 表总分片数
|
||||
*/
|
||||
@Value("${table.shards}")
|
||||
private int tableShards;
|
||||
|
||||
@Override
|
||||
public void onMessage(String body) {
|
||||
log.info("## TodayNoteLikeIncrementData2DBConsumer 消费到了 MQ: {}", body);
|
||||
// 1. 布隆过滤器判断该日增量数据是否已经记录
|
||||
// Json字符串转DTO
|
||||
LikeUnlikeNoteMqDTO noteLikeCountMqDTO = JsonUtils.parseObject(body, LikeUnlikeNoteMqDTO.class);
|
||||
if (Objects.isNull(noteLikeCountMqDTO)) {
|
||||
return;
|
||||
}
|
||||
log.info("## TodayNoteLikeIncrementData2DBConsumer 笔记点赞数据:{}", JsonUtils.toJsonString(noteLikeCountMqDTO));
|
||||
// 获取被点赞或者取消点赞的笔记ID
|
||||
Long noteId = noteLikeCountMqDTO.getNoteId();
|
||||
// 获取点赞或取消点赞的笔记的创建者ID
|
||||
Long noteCreatorId = noteLikeCountMqDTO.getNoteCreatorId();
|
||||
|
||||
// 消息体 JSON 字符串转 DTO
|
||||
LikeUnlikeNoteMqDTO unlikeNoteMqDTO = JsonUtils.parseObject(body, LikeUnlikeNoteMqDTO.class);
|
||||
|
||||
if (Objects.isNull(unlikeNoteMqDTO)) return;
|
||||
|
||||
// 被点赞、取消点赞的笔记 ID
|
||||
Long noteId = unlikeNoteMqDTO.getNoteId();
|
||||
// 笔记的发布者 ID
|
||||
Long noteCreatorId = unlikeNoteMqDTO.getNoteCreatorId();
|
||||
|
||||
// 今日日期
|
||||
String date = LocalDate.now()
|
||||
.format(DateTimeFormatter.ofPattern("yyyyMMdd")); // 转字符串
|
||||
.format(DateTimeFormatter.ofPattern("yyyyMMdd")); // 格式化
|
||||
|
||||
String bloomKey = RedisKeyConstants.buildBloomUserNoteLikeListKey(date);
|
||||
// ------------------------- 笔记的点赞数变更记录 -------------------------
|
||||
// 笔记对应的 Bloom Key
|
||||
String noteBloomKey = RedisKeyConstants.buildBloomUserNoteLikeNoteIdListKey(date);
|
||||
|
||||
// 1. 布隆过滤器判断该日增量数据是否已经记录
|
||||
DefaultRedisScript<Long> script = new DefaultRedisScript<>();
|
||||
@@ -73,36 +72,52 @@ public class TodayNoteLikeIncrementData2DBConsumer implements RocketMQListener<S
|
||||
script.setResultType(Long.class);
|
||||
|
||||
// 执行 Lua 脚本,拿到返回结果
|
||||
Long result = redisTemplate.execute(script, Collections.singletonList(bloomKey), noteId);
|
||||
log.info("布隆过滤器判断结果:{}", result);
|
||||
Long result = redisTemplate.execute(script, Collections.singletonList(noteBloomKey), noteId);
|
||||
|
||||
// Lua 脚本:添加到布隆过滤器
|
||||
RedisScript<Long> bloomAddScript = RedisScript.of("return redis.call('BF.ADD', KEYS[1], ARGV[1])", Long.class);
|
||||
|
||||
// 若布隆过滤器判断不存在(绝对正确)
|
||||
if (Objects.equals(result, 0L)) {
|
||||
// 2. 若无,才会落库,减轻数据库压力
|
||||
// 根据分片总数,取模,分别获取对应的分片序号
|
||||
long userIdHashKey = noteCreatorId % tableShards;
|
||||
long noteIdHashKey = noteId % tableShards;
|
||||
log.info("根据分片总数,取模,分别获取对应的分片序号user:{},note:{}", userIdHashKey, noteIdHashKey);
|
||||
|
||||
// 编程式事务,保证多语句的原子性
|
||||
transactionTemplate.execute(status -> {
|
||||
// 根据分片总数,取模,获取对应的分片序号
|
||||
long noteIdHashKey = noteId % tableShards;
|
||||
|
||||
try {
|
||||
// 将日增量变更数据,分别写入两张表
|
||||
// 将日增量变更数据落库
|
||||
// - t_data_align_note_like_count_temp_日期_分片序号
|
||||
// - t_data_align_user_like_count_temp_日期_分片序号
|
||||
insertRecordMapper.insert2DataAlignNoteLikeCountTempTable(TableConstants.buildTableNameSuffix(date, noteIdHashKey), noteId);
|
||||
insertRecordMapper.insert2DataAlignUserLikeCountTempTable(TableConstants.buildTableNameSuffix(date, userIdHashKey), noteCreatorId);
|
||||
return true;
|
||||
} catch (Exception ex) {
|
||||
status.setRollbackOnly();
|
||||
log.error("## TodayNoteLikeIncrementData2DBConsumer 落库失败,回滚事务", ex);
|
||||
insertMapper.insert2DataAlignNoteLikeCountTempTable(TableConstants.buildTableNameSuffix(date, noteIdHashKey), noteId);
|
||||
} catch (Exception e) {
|
||||
log.error("", e);
|
||||
}
|
||||
return false;
|
||||
});
|
||||
// 3. 数据库写入成功后,再添加布隆过滤器中
|
||||
|
||||
// 4. 数据库写入成功后,再添加布隆过滤器中
|
||||
RedisScript<Long> bloomAddScript = RedisScript.of("return redis.call('BF.ADD', KEYS[1], ARGV[1])", Long.class);
|
||||
redisTemplate.execute(bloomAddScript, Collections.singletonList(bloomKey), noteId);
|
||||
redisTemplate.execute(bloomAddScript, Collections.singletonList(noteBloomKey), noteId);
|
||||
}
|
||||
|
||||
// ------------------------- 笔记发布者获得的点赞数变更记录 -------------------------
|
||||
// 笔记发布者对应的 Bloom Key
|
||||
String userBloomKey = RedisKeyConstants.buildBloomUserNoteLikeUserIdListKey(date);
|
||||
// 执行 Lua 脚本,拿到返回结果
|
||||
result = redisTemplate.execute(script, Collections.singletonList(userBloomKey), noteCreatorId);
|
||||
// 若布隆过滤器判断不存在(绝对正确)
|
||||
if (Objects.equals(result, 0L)) {
|
||||
// 2. 若无,才会落库,减轻数据库压力
|
||||
|
||||
// 根据分片总数,取模,获取对应的分片序号
|
||||
long userIdHashKey = noteCreatorId % tableShards;
|
||||
|
||||
try {
|
||||
// 将日增量变更数据落库
|
||||
// - t_data_align_user_like_count_temp_日期_分片序号
|
||||
insertMapper.insert2DataAlignUserLikeCountTempTable(TableConstants.buildTableNameSuffix(date, userIdHashKey), noteCreatorId);
|
||||
} catch (Exception e) {
|
||||
log.error("", e);
|
||||
}
|
||||
|
||||
// 4. 数据库写入成功后,再添加布隆过滤器中
|
||||
redisTemplate.execute(bloomAddScript, Collections.singletonList(userBloomKey), noteCreatorId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.InsertMapper;
|
||||
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 InsertMapper insertMapper;
|
||||
|
||||
/**
|
||||
* 表总分片数
|
||||
*/
|
||||
@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_日期_分片序号
|
||||
insertMapper.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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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.InsertMapper;
|
||||
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 InsertMapper insertMapper;
|
||||
|
||||
/**
|
||||
* 表总分片数
|
||||
*/
|
||||
@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_日期_分片序号
|
||||
insertMapper.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_日期_分片序号
|
||||
insertMapper.insert2DataAlignUserFansCountTempTable(
|
||||
TableConstants.buildTableNameSuffix(date, targetUserIdHashKey), targetUserId);
|
||||
} catch (Exception e) {
|
||||
log.error("", e);
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
|
||||
// 数据库写入成功后,再添加布隆过滤器中
|
||||
redisTemplate.execute(bloomAddScript, Collections.singletonList(targetUserBloomKey), targetUserId);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
package com.hanserwei.hannote.data.align.domain.mapper;
|
||||
|
||||
import org.apache.ibatis.annotations.Param;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
public interface DeleteRecordMapper {
|
||||
|
||||
/**
|
||||
* 日增量表:关注数计数变更 - 批量删除
|
||||
*
|
||||
* @param userIds 用户 ID
|
||||
*/
|
||||
void batchDeleteDataAlignFollowingCountTempTable(@Param("tableNameSuffix") String tableNameSuffix,
|
||||
@Param("userIds") List<Long> userIds);
|
||||
}
|
||||
@@ -5,7 +5,7 @@ import org.apache.ibatis.annotations.Param;
|
||||
/**
|
||||
* 添加记录
|
||||
*/
|
||||
public interface InsertRecordMapper {
|
||||
public interface InsertMapper {
|
||||
|
||||
/**
|
||||
* 笔记点赞数:计数变更
|
||||
@@ -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);
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
package com.hanserwei.hannote.data.align.domain.mapper;
|
||||
|
||||
import org.apache.ibatis.annotations.Param;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 查询
|
||||
*/
|
||||
public interface SelectRecordMapper {
|
||||
|
||||
|
||||
/**
|
||||
* 日增量表:关注数计数变更 - 批量查询
|
||||
*
|
||||
* @param tableNameSuffix 表名后缀
|
||||
* @param batchSize 批量大小
|
||||
* @return 批量查询结果
|
||||
*/
|
||||
List<Long> selectBatchFromDataAlignFollowingCountTempTable(@Param("tableNameSuffix") String tableNameSuffix,
|
||||
@Param("batchSize") int batchSize);
|
||||
|
||||
/**
|
||||
* 查询 t_following 关注表,获取关注总数
|
||||
*
|
||||
* @param userId 用户 ID
|
||||
* @return 关注总数
|
||||
*/
|
||||
int selectCountFromFollowingTableByUserId(long userId);
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
package com.hanserwei.hannote.data.align.domain.mapper;
|
||||
|
||||
import org.apache.ibatis.annotations.Param;
|
||||
|
||||
public interface UpdateRecordMapper {
|
||||
|
||||
/**
|
||||
* 更新 t_user_count 计数表总关注数
|
||||
*
|
||||
* @param userId 用户 ID
|
||||
* @return 更新行数
|
||||
*/
|
||||
int updateUserFollowingTotalByUserId(@Param("userId") long userId,
|
||||
@Param("followingTotal") int followingTotal);
|
||||
}
|
||||
@@ -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,7 +47,8 @@ public class CreateTableXxlJob {
|
||||
// 表名后缀
|
||||
String tableNameSuffix = TableConstants.buildTableNameSuffix(date, hashKey);
|
||||
|
||||
// 创建表
|
||||
transactionTemplate.execute(status -> {
|
||||
try {
|
||||
// 创建表
|
||||
createTableMapper.createDataAlignFollowingCountTempTable(tableNameSuffix);
|
||||
createTableMapper.createDataAlignFansCountTempTable(tableNameSuffix);
|
||||
@@ -50,6 +57,14 @@ public class CreateTableXxlJob {
|
||||
createTableMapper.createDataAlignUserLikeCountTempTable(tableNameSuffix);
|
||||
createTableMapper.createDataAlignNoteLikeCountTempTable(tableNameSuffix);
|
||||
createTableMapper.createDataAlignNotePublishCountTempTable(tableNameSuffix);
|
||||
return true;
|
||||
} catch (Exception e) {
|
||||
status.setRollbackOnly();
|
||||
log.error("创建表失败", e);
|
||||
}
|
||||
return false;
|
||||
});
|
||||
|
||||
}
|
||||
}
|
||||
XxlJobHelper.log("## 创建日增量数据表成功,表名后缀: {}...", date);
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
package com.hanserwei.hannote.data.align.job;
|
||||
|
||||
import cn.hutool.core.collection.CollUtil;
|
||||
import com.hanserwei.hannote.data.align.constant.RedisKeyConstants;
|
||||
import com.hanserwei.hannote.data.align.constant.TableConstants;
|
||||
import com.hanserwei.hannote.data.align.domain.mapper.DeleteRecordMapper;
|
||||
import com.hanserwei.hannote.data.align.domain.mapper.SelectRecordMapper;
|
||||
import com.hanserwei.hannote.data.align.domain.mapper.UpdateRecordMapper;
|
||||
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.data.redis.core.RedisTemplate;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.time.LocalDate;
|
||||
import java.time.format.DateTimeFormatter;
|
||||
import java.util.List;
|
||||
|
||||
@Component
|
||||
@Slf4j
|
||||
public class FollowingCountShardingXxlJob {
|
||||
|
||||
@Resource
|
||||
private SelectRecordMapper selectRecordMapper;
|
||||
@Resource
|
||||
private UpdateRecordMapper updateRecordMapper;
|
||||
@Resource
|
||||
private DeleteRecordMapper deleteRecordMapper;
|
||||
@Resource
|
||||
private RedisTemplate<String, Object> redisTemplate;
|
||||
|
||||
/**
|
||||
* 分片广播任务
|
||||
*/
|
||||
@XxlJob("followingCountShardingJobHandler")
|
||||
public void followingCountShardingJobHandler() {
|
||||
// 获取分片参数
|
||||
// 分片序号
|
||||
int shardIndex = XxlJobHelper.getShardIndex();
|
||||
// 分片总数
|
||||
int shardTotal = XxlJobHelper.getShardTotal();
|
||||
|
||||
XxlJobHelper.log("=================> 开始定时分片广播任务:对当日发生变更的用户关注数进行对齐");
|
||||
XxlJobHelper.log("分片参数:当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);
|
||||
|
||||
log.info("分片参数:当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);
|
||||
|
||||
// 表后缀
|
||||
String date = LocalDate.now().minusDays(1) // 昨日的日期
|
||||
.format(DateTimeFormatter.ofPattern("yyyyMMdd")); // 转字符串
|
||||
// 表名后缀
|
||||
String tableNameSuffix = TableConstants.buildTableNameSuffix(date, shardIndex);
|
||||
|
||||
// 一批次 1000 条
|
||||
int batchSize = 1000;
|
||||
// 共对齐了多少条记录,默认为 0
|
||||
int processedTotal = 0;
|
||||
|
||||
while (true) {
|
||||
// 1. 分批次查询,如一次查询1000 条,直到查询完毕
|
||||
List<Long> userIds = selectRecordMapper.selectBatchFromDataAlignFollowingCountTempTable(tableNameSuffix, batchSize);
|
||||
|
||||
// 若记录为空,则结束循环
|
||||
if (CollUtil.isEmpty(userIds)) {
|
||||
break;
|
||||
}
|
||||
|
||||
// 循环这一批发生变更的用户 ID
|
||||
userIds.forEach(userId -> {
|
||||
// 2: 对 t_following 关注表执行 count(*) 操作,获取关注总数
|
||||
int followingTotal = selectRecordMapper.selectCountFromFollowingTableByUserId(userId);
|
||||
|
||||
// 3: 更新 t_user_count 表
|
||||
int count = updateRecordMapper.updateUserFollowingTotalByUserId(userId, followingTotal);
|
||||
// 更新对应 Redis 缓存
|
||||
if (count > 0) {
|
||||
String redisKey = RedisKeyConstants.buildCountUserKey(userId);
|
||||
// 判断 Hash 是否存在
|
||||
boolean hashKey = redisTemplate.hasKey(redisKey);
|
||||
// 若存在
|
||||
if (hashKey) {
|
||||
// 更新 Hash 中的 Field 关注总数
|
||||
redisTemplate.opsForHash().put(redisKey, RedisKeyConstants.FIELD_FOLLOWING_TOTAL, followingTotal);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
// 4. 批量物理删除这一批次记录
|
||||
|
||||
deleteRecordMapper.batchDeleteDataAlignFollowingCountTempTable(tableNameSuffix, userIds);
|
||||
|
||||
// 当前已处理的记录数
|
||||
processedTotal += userIds.size();
|
||||
}
|
||||
|
||||
XxlJobHelper.log("=================> 结束定时分片广播任务:对当日发生变更的用户关注数进行对齐,共对齐记录数:{}", processedTotal);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
}
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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 ='数据对齐日增量表:用户发布笔记数';
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
<?xml version="1.0" encoding="UTF-8" ?>
|
||||
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
|
||||
<mapper namespace="com.hanserwei.hannote.data.align.domain.mapper.DeleteRecordMapper">
|
||||
<delete id="batchDeleteDataAlignFollowingCountTempTable" parameterType="list">
|
||||
delete
|
||||
from `t_data_align_following_count_temp_${tableNameSuffix}`
|
||||
where user_id in
|
||||
<foreach collection="userIds" open="(" item="userId" close=")" separator=",">
|
||||
#{userId}
|
||||
</foreach>
|
||||
</delete>
|
||||
</mapper>
|
||||
@@ -1,6 +1,6 @@
|
||||
<?xml version="1.0" encoding="UTF-8" ?>
|
||||
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
|
||||
<mapper namespace="com.hanserwei.hannote.data.align.domain.mapper.InsertRecordMapper">
|
||||
<mapper namespace="com.hanserwei.hannote.data.align.domain.mapper.InsertMapper">
|
||||
<insert id="insert2DataAlignNoteLikeCountTempTable" parameterType="map">
|
||||
insert into `t_data_align_note_like_count_temp_${tableNameSuffix}` (note_id)
|
||||
values (#{noteId})
|
||||
@@ -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>
|
||||
@@ -0,0 +1,16 @@
|
||||
<?xml version="1.0" encoding="UTF-8" ?>
|
||||
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
|
||||
<mapper namespace="com.hanserwei.hannote.data.align.domain.mapper.SelectRecordMapper">
|
||||
<select id="selectBatchFromDataAlignFollowingCountTempTable" resultType="long" parameterType="map">
|
||||
select user_id
|
||||
from `t_data_align_following_count_temp_${tableNameSuffix}`
|
||||
order by id
|
||||
limit #{batchSize}
|
||||
</select>
|
||||
|
||||
<select id="selectCountFromFollowingTableByUserId" parameterType="map" resultType="int">
|
||||
select count(*)
|
||||
from t_following
|
||||
where user_id = #{userId}
|
||||
</select>
|
||||
</mapper>
|
||||
@@ -0,0 +1,9 @@
|
||||
<?xml version="1.0" encoding="UTF-8" ?>
|
||||
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
|
||||
<mapper namespace="com.hanserwei.hannote.data.align.domain.mapper.UpdateRecordMapper">
|
||||
<update id="updateUserFollowingTotalByUserId" parameterType="map">
|
||||
update t_user_count
|
||||
set following_total = #{followingTotal}
|
||||
where user_id = #{userId}
|
||||
</update>
|
||||
</mapper>
|
||||
@@ -651,6 +651,7 @@ public class NoteServiceImpl extends ServiceImpl<NoteDOMapper, NoteDO> implement
|
||||
//从数据库中校验笔记是否被点赞,并异步初始化布隆过滤器,设置过期时间
|
||||
long count = noteLikeDOService.count(new LambdaQueryWrapper<>(NoteLikeDO.class)
|
||||
.eq(NoteLikeDO::getNoteId, noteId)
|
||||
.eq(NoteLikeDO::getUserId, userId)
|
||||
.eq(NoteLikeDO::getStatus, LikeStatusEnum.LIKE.getCode()));
|
||||
|
||||
// 保底1天+随机秒数
|
||||
|
||||
@@ -77,8 +77,8 @@ Authorization: Bearer {{token}}
|
||||
"imgUris": [
|
||||
"https://cdn.pixabay.com/photo/2025/10/05/15/06/autumn-9875155_1280.jpg"
|
||||
],
|
||||
"title": "第三篇图文笔记",
|
||||
"content": "这个是第三篇图文笔记的测试",
|
||||
"title": "bug修复2",
|
||||
"content": "bugbugbug",
|
||||
"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
|
||||
}
|
||||
|
||||
### 取消关注
|
||||
@@ -197,16 +197,16 @@ Authorization: Bearer {{token}}
|
||||
### 笔记点赞入口
|
||||
POST http://localhost:8000/note/note/like
|
||||
Content-Type: application/json
|
||||
Authorization: Bearer {{token}}
|
||||
Authorization: Bearer {{thirdToken}}
|
||||
|
||||
{
|
||||
"id": 1977249693272375330
|
||||
"id": 1981698494959714362
|
||||
}
|
||||
|
||||
### 笔记取消点赞入口
|
||||
POST http://localhost:8000/note/note/unlike
|
||||
Content-Type: application/json
|
||||
Authorization: Bearer {{token}}
|
||||
Authorization: Bearer {{otherToken}}
|
||||
|
||||
{
|
||||
"id": 1977249693272375330
|
||||
@@ -215,16 +215,16 @@ Authorization: Bearer {{token}}
|
||||
### 笔记收藏入口
|
||||
POST http://localhost:8000/note/note/collect
|
||||
Content-Type: application/json
|
||||
Authorization: Bearer {{token}}
|
||||
Authorization: Bearer {{thirdToken}}
|
||||
|
||||
{
|
||||
"id": 1977249693272375330
|
||||
"id": 1981698494959714362
|
||||
}
|
||||
|
||||
### 笔记取消收藏入口
|
||||
POST http://localhost:8000/note/note/uncollect
|
||||
Content-Type: application/json
|
||||
Authorization: Bearer {{token}}
|
||||
Authorization: Bearer {{otherToken}}
|
||||
|
||||
{
|
||||
"id": 1977249693272375330
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
"dev": {
|
||||
"token": "4bXpiBbjXEDFE4ZpqjCOHu1rP81qepl2ROOygrxRGb61K536ckLuyAwfyQHSMcyRdUzf8CxntLEMfbU2ynbYx9nJKlx4vpWZrHqv2mI4iMhnShQ4mPBi7OPPgZi22O2f",
|
||||
"otherToken": "mqFNHrWkPcipIAvw7Gn4cigOWYP54sn8HYlQX3CXTxHf90DhjFiROhWVgPqLBi35xKXOOfHlXeEdaQrkXf1JXd8hbXBOdZqnrycW96BJwTbUS40EqIZifVgPun3ai0Ek",
|
||||
"thirdToken": "iA8XE1vFDXYwgNPnRyIrNaj5EKcQypUTtn91wCMGtF8FfFdFzvRUad4Q7shLkOgUQ5QMB5n25JP91vpYIvr7udoL1HUxdjlSlXCEXivTQlgaABkz5owdhzhyHqGg0XP8",
|
||||
"noteId": "1977249693272375330",
|
||||
"userId": "100",
|
||||
"otherUserId": "2100"
|
||||
|
||||
Reference in New Issue
Block a user