Toyas/HAOVP
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(content): Improve video statistics messaging fallback

Browse Source
This commit is contained in:
Toyas
2025-04-10 04:14:49 +08:00
parent d1b55d770b
commit 2f49bed861
4 changed files with 77 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video-content/src/main/java/com/nuaa/video/content/config/RocketConfig.java
+15 -2
View File
@@ -1,11 +1,11 @@
package com.nuaa.video.content.config;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
@Configuration
public class RocketConfig {
@@ -22,8 +22,21 @@ public class RocketConfig {
DefaultMQProducer defaultMQProducer = new DefaultMQProducer(producerGroup);
defaultMQProducer.setProducerGroup(producerGroup);
defaultMQProducer.setNamesrvAddr(nameServer);
// 设置消息最大长度
defaultMQProducer.setMaxMessageSize(4 * 1024 * 1024);
// 启用自动创建主题（注意：这需要broker端支持）
defaultMQProducer.setCreateTopicKey("AUTO_CREATE_TOPIC_KEY");
defaultMQProducer.setSendMsgTimeout(10000);
rocketMQTemplate.setProducer(defaultMQProducer);
// 创建并设置消息转换器
MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
converter.setSerializedPayloadClass(String.class);
converter.setStrictContentTypeMatch(false);
rocketMQTemplate.setMessageConverter(converter);
return rocketMQTemplate;
}
}
video-content/src/main/java/com/nuaa/video/content/mq/producer/VideoMessageProducer.java
+54 -8
View File
@@ -1,5 +1,9 @@
package com.nuaa.video.content.mq.producer;
import com.nuaa.video.content.mapper.ContentVideoMapper;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
@@ -8,6 +12,8 @@ import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import java.io.Serializable;
/**
* 视频消息生产者
*/
@@ -17,6 +23,7 @@ import org.springframework.stereotype.Component;
public class VideoMessageProducer {
private final RocketMQTemplate rocketMQTemplate;
private final ContentVideoMapper videoMapper;
// 定义标准化的主题名称，添加业务域前缀
private static final String TOPIC_PREFIX = "video-platform-";
@@ -89,25 +96,64 @@ public class VideoMessageProducer {
public void sendVideoStatsMessage(VideoStatsMessage message) {
log.info("准备发送视频统计消息: videoId={}, userId={}, action={}",
message.getVideoId(), message.getUserId(), message.getAction());
try {
SendResult sendResult = rocketMQTemplate.syncSend(VIDEO_STATS_TOPIC,
MessageBuilder.withPayload(message).build());
log.info("视频统计消息发送成功: videoId={}, action={}, msgId={}",
message.getVideoId(), message.getAction(), sendResult.getMsgId());
// 使用异步发送代替同步发送，避免阻塞主线程
rocketMQTemplate.asyncSend(VIDEO_STATS_TOPIC,
MessageBuilder.withPayload(message).build(),
new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
log.info("视频统计消息发送成功: videoId={}, action={}, msgId={}",
message.getVideoId(), message.getAction(), sendResult.getMsgId());
}
@Override
public void onException(Throwable throwable) {
log.error("视频统计消息发送失败: videoId={}, action={}",
message.getVideoId(), message.getAction(), throwable);
// 降级处理：直接更新数据库
updateVideoStatsDirectly(message);
}
});
} catch (Exception e) {
log.error("视频统计消息发送失败: videoId={}, action={}",
log.error("视频统计消息发送异常，降级处理: videoId={}, action={}",
message.getVideoId(), message.getAction(), e);
// 降级处理：直接更新数据库
updateVideoStatsDirectly(message);
}
}
private void updateVideoStatsDirectly(VideoStatsMessage message) {
try {
Long videoId = message.getVideoId();
String action = message.getAction();
Integer count = message.getCount() == null ? 1 : message.getCount();
// 直接调用数据库更新
switch (action) {
case "play":
videoMapper.increaseViewCount(videoId, count);
log.info("直接更新数据库：视频播放计数 +{}, videoId={}", count, videoId);
break;
// 其他case处理...
}
} catch (Exception e) {
log.error("降级处理失败: 直接更新数据库出错", e);
}
}
/**
* 视频统计消息
*/
@lombok.Data
public static class VideoStatsMessage {
@NoArgsConstructor
@AllArgsConstructor
@Data
public static class VideoStatsMessage implements Serializable {
private static final long serialVersionUID = 1L;
private Long videoId;
private Long userId;
private String action; // play, like, favorite, coin, share, danmaku
private Integer count;
private Integer count = 1;
}
}
video-content/src/main/java/com/nuaa/video/content/service/impl/VideoServiceImpl.java
+8 -10
View File
@@ -196,18 +196,16 @@ public class VideoServiceImpl extends ServiceImpl<ContentVideoMapper, ContentVid
int coinCount = videoCoinMapper.getUserCoinCount(userId, id);
videoInfoVO.setIsCoined(coinCount > 0);
// 增加播放数
videoMessageProducer.sendVideoStatsMessage(
new VideoMessageProducer.VideoStatsMessage() {{
setVideoId(id);
setUserId(userId);
setAction("play");
setCount(1);
}}
);
}
// 增加播放数
VideoMessageProducer.VideoStatsMessage message = new VideoMessageProducer.VideoStatsMessage();
message.setVideoId(id);
message.setUserId(userId);
message.setAction("play");
message.setCount(1);
videoMessageProducer.sendVideoStatsMessage(message);
return Result.success(videoInfoVO);
}