diff --git a/han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/domain/HannoteCountBizApplication.java b/han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/HannoteCountBizApplication.java similarity index 89% rename from han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/domain/HannoteCountBizApplication.java rename to han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/HannoteCountBizApplication.java index a2e4cc9..22c284c 100644 --- a/han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/domain/HannoteCountBizApplication.java +++ b/han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/HannoteCountBizApplication.java @@ -1,4 +1,4 @@ -package com.hanserwei.hannote.count.biz.domain; +package com.hanserwei.hannote.count.biz; import org.mybatis.spring.annotation.MapperScan; import org.springframework.boot.SpringApplication; diff --git a/han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/constant/MQConstants.java b/han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/constant/MQConstants.java new file mode 100644 index 0000000..b37023b --- /dev/null +++ b/han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/constant/MQConstants.java @@ -0,0 +1,15 @@ +package com.hanserwei.hannote.count.biz.constant; + +public interface MQConstants { + + /** + * Topic: 关注数计数 + */ + String TOPIC_COUNT_FOLLOWING = "CountFollowingTopic"; + + /** + * Topic: 粉丝数计数 + */ + String TOPIC_COUNT_FANS = "CountFansTopic"; + +} \ No newline at end of file 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 new file mode 100644 index 0000000..2e1e149 --- /dev/null +++ b/han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/consumer/CountFansConsumer.java @@ -0,0 +1,20 @@ +package com.hanserwei.hannote.count.biz.consumer; + +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; + +@Component +@RocketMQMessageListener( + consumerGroup = "han_note_group_" + MQConstants.TOPIC_COUNT_FANS, + topic = MQConstants.TOPIC_COUNT_FANS +) +@Slf4j +public class CountFansConsumer implements RocketMQListener { + @Override + public void onMessage(String body) { + log.info("## 消费了 MQ [计数：粉丝数]: {}", body); + } +} diff --git a/han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/consumer/CountFollowingConsumer.java b/han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/consumer/CountFollowingConsumer.java new file mode 100644 index 0000000..3bff007 --- /dev/null +++ b/han-note-count/han-note-count-biz/src/main/java/com/hanserwei/hannote/count/biz/consumer/CountFollowingConsumer.java @@ -0,0 +1,20 @@ +package com.hanserwei.hannote.count.biz.consumer; + +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; + +@Component +@RocketMQMessageListener( + consumerGroup = "han_note_group_" + MQConstants.TOPIC_COUNT_FOLLOWING, + topic = MQConstants.TOPIC_COUNT_FOLLOWING +) +@Slf4j +public class CountFollowingConsumer implements RocketMQListener { + @Override + public void onMessage(String body) { + log.info("## 消费了 MQ [计数：关注数]: {}", body); + } +} diff --git a/han-note-count/han-note-count-biz/src/main/resources/application.yml b/han-note-count/han-note-count-biz/src/main/resources/application.yml index 02c3b36..e387d7f 100644 --- a/han-note-count/han-note-count-biz/src/main/resources/application.yml +++ b/han-note-count/han-note-count-biz/src/main/resources/application.yml @@ -28,7 +28,4 @@ mybatis-plus: log-impl: org.apache.ibatis.logging.slf4j.Slf4jImpl global-config: banner: false - mapper-locations: classpath*:/mapperxml/*.xml -mq-consumer: # MQ 消费者 - follow-unfollow: # 关注、取关 - rate-limit: 5000 # 每秒限流阈值 \ No newline at end of file + mapper-locations: classpath*:/mapperxml/*.xml \ No newline at end of file diff --git a/han-note-user-relation/han-note-user-relation-biz/src/main/java/com/hanserwei/hannote/user/relation/biz/constant/MQConstants.java b/han-note-user-relation/han-note-user-relation-biz/src/main/java/com/hanserwei/hannote/user/relation/biz/constant/MQConstants.java index 3f3423d..49ce2c0 100644 --- a/han-note-user-relation/han-note-user-relation-biz/src/main/java/com/hanserwei/hannote/user/relation/biz/constant/MQConstants.java +++ b/han-note-user-relation/han-note-user-relation-biz/src/main/java/com/hanserwei/hannote/user/relation/biz/constant/MQConstants.java @@ -16,4 +16,14 @@ public interface MQConstants { * 取关标签 */ String TAG_UNFOLLOW = "Unfollow"; + + /** + * Topic: 关注数计数 + */ + String TOPIC_COUNT_FOLLOWING = "CountFollowingTopic"; + + /** + * Topic: 粉丝数计数 + */ + String TOPIC_COUNT_FANS = "CountFansTopic"; } \ No newline at end of file diff --git a/han-note-user-relation/han-note-user-relation-biz/src/main/java/com/hanserwei/hannote/user/relation/biz/consumer/FollowUnfollowConsumer.java b/han-note-user-relation/han-note-user-relation-biz/src/main/java/com/hanserwei/hannote/user/relation/biz/consumer/FollowUnfollowConsumer.java index 6844ab8..26095a0 100644 --- a/han-note-user-relation/han-note-user-relation-biz/src/main/java/com/hanserwei/hannote/user/relation/biz/consumer/FollowUnfollowConsumer.java +++ b/han-note-user-relation/han-note-user-relation-biz/src/main/java/com/hanserwei/hannote/user/relation/biz/consumer/FollowUnfollowConsumer.java @@ -7,6 +7,8 @@ import com.hanserwei.hannote.user.relation.biz.constant.MQConstants; import com.hanserwei.hannote.user.relation.biz.constant.RedisKeyConstants; import com.hanserwei.hannote.user.relation.biz.domain.dataobject.FansDO; import com.hanserwei.hannote.user.relation.biz.domain.dataobject.FollowingDO; +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 com.hanserwei.hannote.user.relation.biz.service.FansDOService; @@ -15,13 +17,17 @@ import com.hanserwei.hannote.user.relation.biz.util.DateUtils; import jakarta.annotation.Resource; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.spring.annotation.ConsumeMode; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; +import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.springframework.core.io.ClassPathResource; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.script.DefaultRedisScript; +import org.springframework.messaging.support.MessageBuilder; import org.springframework.scripting.support.ResourceScriptSource; import org.springframework.stereotype.Component; import org.springframework.transaction.support.TransactionTemplate; @@ -32,8 +38,8 @@ import java.util.Objects; @Component @RocketMQMessageListener( - consumerGroup = "han_note_group_" + MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW, - topic = MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW, + consumerGroup = "han_note_group_" + MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW, //han_note_group_FollowUnfollowTopic + topic = MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW, //FollowUnfollowTopic consumeMode = ConsumeMode.ORDERLY ) @Slf4j @@ -47,6 +53,8 @@ public class FollowUnfollowConsumer implements RocketMQListener { private RateLimiter rateLimiter; @Resource private RedisTemplate redisTemplate; + @Resource + private RocketMQTemplate rocketMQTemplate; @Override public void onMessage(Message message) { @@ -114,6 +122,17 @@ public class FollowUnfollowConsumer implements RocketMQListener { String fansRedisKey = RedisKeyConstants.buildUserFansKey(unfollowUserId); // 删除指定粉丝 redisTemplate.opsForZSet().remove(fansRedisKey, userId); + + // 发送MQ消息通知计数服务，统计关注数 + // 构建DTO对象 + CountFollowUnfollowMqDTO countFollowUnfollowMqDTO = CountFollowUnfollowMqDTO.builder() + .userId(userId) + .targetUserId(unfollowUserId) + .type(FollowUnfollowTypeEnum.UNFOLLOW.getCode()) + .build(); + + // 发送MQ + sendMQ(countFollowUnfollowMqDTO); } } @@ -177,6 +196,53 @@ public class FollowUnfollowConsumer implements RocketMQListener { // 执行Lua脚本 redisTemplate.execute(script, Collections.singletonList(fansZSetKey), userId, timestamp); + + // 发送MQ消息通知计数服务，统计关注数 + // 构建消息体 + CountFollowUnfollowMqDTO countFollowUnfollowMqDTO = CountFollowUnfollowMqDTO.builder() + .userId(userId) + .targetUserId(followUserId) + .type(FollowUnfollowTypeEnum.FOLLOW.getCode()) + .build(); + + sendMQ(countFollowUnfollowMqDTO); } } + + /** + * 发送MQ消息 + * + * @param countFollowUnfollowMqDTO 消息体 + */ + private void sendMQ(CountFollowUnfollowMqDTO countFollowUnfollowMqDTO) { + // 构建MQ消息体 + org.springframework.messaging.Message message = MessageBuilder.withPayload(JsonUtils.toJsonString(countFollowUnfollowMqDTO)) + .build(); + + // 异步发送 MQ 消息 + rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_FOLLOWING, message, new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + log.info("==> 【计数服务：关注数】MQ 发送成功，SendResult: {}", sendResult); + } + + @Override + public void onException(Throwable throwable) { + log.error("==> 【计数服务：关注数】MQ 发送异常: ", throwable); + } + }); + + // 发送 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); + } + }); + } } diff --git a/han-note-user-relation/han-note-user-relation-biz/src/main/java/com/hanserwei/hannote/user/relation/biz/enums/FollowUnfollowTypeEnum.java b/han-note-user-relation/han-note-user-relation-biz/src/main/java/com/hanserwei/hannote/user/relation/biz/enums/FollowUnfollowTypeEnum.java new file mode 100644 index 0000000..7ed572d --- /dev/null +++ b/han-note-user-relation/han-note-user-relation-biz/src/main/java/com/hanserwei/hannote/user/relation/biz/enums/FollowUnfollowTypeEnum.java @@ -0,0 +1,17 @@ +package com.hanserwei.hannote.user.relation.biz.enums; + +import lombok.AllArgsConstructor; +import lombok.Getter; + +@Getter +@AllArgsConstructor +public enum FollowUnfollowTypeEnum { + // 关注 + FOLLOW(1), + // 取关 + UNFOLLOW(0), + ; + + private final Integer code; + +} \ No newline at end of file diff --git a/han-note-user-relation/han-note-user-relation-biz/src/main/java/com/hanserwei/hannote/user/relation/biz/model/dto/CountFollowUnfollowMqDTO.java b/han-note-user-relation/han-note-user-relation-biz/src/main/java/com/hanserwei/hannote/user/relation/biz/model/dto/CountFollowUnfollowMqDTO.java new file mode 100644 index 0000000..ca0edc8 --- /dev/null +++ b/han-note-user-relation/han-note-user-relation-biz/src/main/java/com/hanserwei/hannote/user/relation/biz/model/dto/CountFollowUnfollowMqDTO.java @@ -0,0 +1,29 @@ +package com.hanserwei.hannote.user.relation.biz.model.dto; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +@Data +@AllArgsConstructor +@NoArgsConstructor +@Builder +public class CountFollowUnfollowMqDTO { + + /** + * 原用户 + */ + private Long userId; + + /** + * 目标用户 + */ + private Long targetUserId; + + /** + * 1:关注 0:取关 + */ + private Integer type; + +} \ No newline at end of file