diff --git a/.idea/.cache/.Apifox_Helper/.toolWindow.db b/.idea/.cache/.Apifox_Helper/.toolWindow.db index fd168ed..aa3ef4c 100644 Binary files a/.idea/.cache/.Apifox_Helper/.toolWindow.db and b/.idea/.cache/.Apifox_Helper/.toolWindow.db differ diff --git a/video-content/src/main/java/com/nuaa/video/content/config/RocketConfig.java b/video-content/src/main/java/com/nuaa/video/content/config/RocketConfig.java index 286df2d..ea86eab 100644 --- a/video-content/src/main/java/com/nuaa/video/content/config/RocketConfig.java +++ b/video-content/src/main/java/com/nuaa/video/content/config/RocketConfig.java @@ -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; } - } \ No newline at end of file diff --git a/video-content/src/main/java/com/nuaa/video/content/mq/producer/VideoMessageProducer.java b/video-content/src/main/java/com/nuaa/video/content/mq/producer/VideoMessageProducer.java index 965e8f3..2ac6af0 100644 --- a/video-content/src/main/java/com/nuaa/video/content/mq/producer/VideoMessageProducer.java +++ b/video-content/src/main/java/com/nuaa/video/content/mq/producer/VideoMessageProducer.java @@ -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; } } \ No newline at end of file diff --git a/video-content/src/main/java/com/nuaa/video/content/service/impl/VideoServiceImpl.java b/video-content/src/main/java/com/nuaa/video/content/service/impl/VideoServiceImpl.java index e2264f1..54ed2b5 100644 --- a/video-content/src/main/java/com/nuaa/video/content/service/impl/VideoServiceImpl.java +++ b/video-content/src/main/java/com/nuaa/video/content/service/impl/VideoServiceImpl.java @@ -196,18 +196,16 @@ public class VideoServiceImpl extends ServiceImpl 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); }