refactor(mq): Optimize RocketMQ topics and message publishing

video-content/src/main/java/com/nuaa/video/content/mq/consumer/VideoMessageConsumer.java

@@ -34,7 +34,7 @@ public class VideoMessageConsumer {
      */
     @Component
     @RocketMQMessageListener(
-            topic = "video-stats-topic",
+            topic = "${rocket-mq.topic-prefix:video-platform-}video-stats",
             consumerGroup = "video-stats-consumer-group",
             consumeMode = ConsumeMode.CONCURRENTLY
     )
@@ -154,13 +154,15 @@ public class VideoMessageConsumer {
      */
     @Component
     @RocketMQMessageListener(
-            topic = "video-process-topic",
+            topic = "${rocket-mq.topic-prefix:video-platform-}video-process",
             consumerGroup = "video-process-consumer-group",
             consumeMode = ConsumeMode.CONCURRENTLY
     )
     @RequiredArgsConstructor
     class VideoProcessListener implements RocketMQListener<Long> {
 
+        private final VideoMessageProducer videoMessageProducer; // 依赖注入
+
         @Override
         @Transactional(rollbackFor = Exception.class)
         public void onMessage(Long videoId) {
@@ -182,7 +184,6 @@ public class VideoMessageConsumer {
                 videoMapper.updateById(video);
 
                 // 发送审核消息
-                VideoMessageProducer videoMessageProducer = new VideoMessageProducer(null);
                 videoMessageProducer.sendVideoReviewMessage(videoId);
 
             } catch (Exception e) {
@@ -197,7 +198,7 @@ public class VideoMessageConsumer {
      */
     @Component
     @RocketMQMessageListener(
-            topic = "video-review-topic",
+            topic = "${rocket-mq.topic-prefix:video-platform-}video-review",
             consumerGroup = "video-review-consumer-group",
             consumeMode = ConsumeMode.CONCURRENTLY
     )

video-content/src/main/java/com/nuaa/video/content/mq/producer/VideoMessageProducer.java

@@ -2,6 +2,8 @@ package com.nuaa.video.content.mq.producer;
 
 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.spring.core.RocketMQTemplate;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.stereotype.Component;
@@ -16,10 +18,12 @@ public class VideoMessageProducer {
 
     private final RocketMQTemplate rocketMQTemplate;
 
-    private static final String VIDEO_UPLOAD_TOPIC = "video-upload-topic";
+    // 定义标准化的主题名称，添加业务域前缀
-    private static final String VIDEO_PROCESS_TOPIC = "video-process-topic";
+    private static final String TOPIC_PREFIX = "video-platform-";
-    private static final String VIDEO_REVIEW_TOPIC = "video-review-topic";
+    private static final String VIDEO_UPLOAD_TOPIC = TOPIC_PREFIX + "video-upload";
-    private static final String VIDEO_STATS_TOPIC = "video-stats-topic";
+    private static final String VIDEO_PROCESS_TOPIC = TOPIC_PREFIX + "video-process";
+    private static final String VIDEO_REVIEW_TOPIC = TOPIC_PREFIX + "video-review";
+    private static final String VIDEO_STATS_TOPIC = TOPIC_PREFIX + "video-stats";
 
     /**
      * 发送视频上传消息
@@ -27,8 +31,23 @@ public class VideoMessageProducer {
      * @param videoId 视频ID
      */
     public void sendVideoUploadMessage(Long videoId) {
-        log.info("发送视频上传消息: videoId={}", videoId);
+        log.info("准备发送视频上传消息: videoId={}", videoId);
-        rocketMQTemplate.convertAndSend(VIDEO_UPLOAD_TOPIC, videoId);
+        try {
+            rocketMQTemplate.asyncSend(VIDEO_UPLOAD_TOPIC, videoId, new SendCallback() {
+                @Override
+                public void onSuccess(SendResult sendResult) {
+                    log.info("视频上传消息发送成功: videoId={}, msgId={}", videoId, sendResult.getMsgId());
+                }
+
+                @Override
+                public void onException(Throwable throwable) {
+                    log.error("视频上传消息发送失败: videoId={}", videoId, throwable);
+                    // 实际应用中可以添加重试逻辑或写入失败队列
+                }
+            });
+        } catch (Exception e) {
+            log.error("视频上传消息发送异常: videoId={}", videoId, e);
+        }
     }
 
     /**
@@ -37,8 +56,14 @@ public class VideoMessageProducer {
      * @param videoId 视频ID
      */
     public void sendVideoProcessMessage(Long videoId) {
-        log.info("发送视频处理消息: videoId={}", videoId);
+        log.info("准备发送视频处理消息: videoId={}", videoId);
-        rocketMQTemplate.convertAndSend(VIDEO_PROCESS_TOPIC, videoId);
+        try {
+            SendResult sendResult = rocketMQTemplate.syncSend(VIDEO_PROCESS_TOPIC, videoId);
+            log.info("视频处理消息发送成功: videoId={}, msgId={}", videoId, sendResult.getMsgId());
+        } catch (Exception e) {
+            log.error("视频处理消息发送失败: videoId={}", videoId, e);
+            // 可以添加重试逻辑或降级处理
+        }
     }
 
     /**
@@ -47,8 +72,13 @@ public class VideoMessageProducer {
      * @param videoId 视频ID
      */
     public void sendVideoReviewMessage(Long videoId) {
-        log.info("发送视频审核消息: videoId={}", videoId);
+        log.info("准备发送视频审核消息: videoId={}", videoId);
-        rocketMQTemplate.convertAndSend(VIDEO_REVIEW_TOPIC, videoId);
+        try {
+            SendResult sendResult = rocketMQTemplate.syncSend(VIDEO_REVIEW_TOPIC, videoId);
+            log.info("视频审核消息发送成功: videoId={}, msgId={}", videoId, sendResult.getMsgId());
+        } catch (Exception e) {
+            log.error("视频审核消息发送失败: videoId={}", videoId, e);
+        }
     }
 
     /**
@@ -57,9 +87,17 @@ public class VideoMessageProducer {
      * @param message 统计消息
      */
     public void sendVideoStatsMessage(VideoStatsMessage message) {
-        log.info("发送视频统计消息: videoId={}, userId={}, action={}",
+        log.info("准备发送视频统计消息: videoId={}, userId={}, action={}",
                 message.getVideoId(), message.getUserId(), message.getAction());
-        rocketMQTemplate.send(VIDEO_STATS_TOPIC, MessageBuilder.withPayload(message).build());
+        try {
+            SendResult sendResult = rocketMQTemplate.syncSend(VIDEO_STATS_TOPIC,
+                    MessageBuilder.withPayload(message).build());
+            log.info("视频统计消息发送成功: videoId={}, action={}, msgId={}",
+                    message.getVideoId(), message.getAction(), sendResult.getMsgId());
+        } catch (Exception e) {
+            log.error("视频统计消息发送失败: videoId={}, action={}",
+                    message.getVideoId(), message.getAction(), e);
+        }
     }
 
     /**

video-content/src/main/resources/application.yml

@@ -62,6 +62,7 @@ spring:
 
   # RocketMQ配置
   rocketmq:
+    topic-prefix: video-platform-
     name-server: ${ROCKETMQ_HOST:5.180.253.115}:${ROCKETMQ_PORT:9876}
     producer:
       group: video-content-producer

video-gateway/src/main/java/com/nuaa/video/gateway/config/LoadBalancerConfig.java

@@ -0,0 +1,16 @@
+package com.nuaa.video.gateway.config;
+
+import org.springframework.cloud.client.loadbalancer.LoadBalanced;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.reactive.function.client.WebClient;
+
+@Configuration
+public class LoadBalancerConfig {
+
+    @Bean
+    @LoadBalanced
+    public WebClient.Builder loadBalancedWebClientBuilder() {
+        return WebClient.builder();
+    }
+}

video-gateway/src/main/resources/application.yml

@@ -65,7 +65,7 @@ spring:
           predicates:
             - Path=/admin/**
           filters:
-            - StripPrefix=1
+            - StripPrefix=0
     sentinel:
       transport:
         dashboard: ${SENTINEL_DASHBOARD_HOST:5.180.253.115}:${SENTINEL_DASHBOARD_PORT:8080}