diff --git a/video-content/src/main/java/com/nuaa/video/content/mq/consumer/VideoMessageConsumer.java b/video-content/src/main/java/com/nuaa/video/content/mq/consumer/VideoMessageConsumer.java index 9e41a02..6efc5fc 100644 --- a/video-content/src/main/java/com/nuaa/video/content/mq/consumer/VideoMessageConsumer.java +++ b/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 { + 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 ) 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 93e8367..965e8f3 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 @@ -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 VIDEO_REVIEW_TOPIC = "video-review-topic"; - private static final String VIDEO_STATS_TOPIC = "video-stats-topic"; + // 定义标准化的主题名称，添加业务域前缀 + private static final String TOPIC_PREFIX = "video-platform-"; + private static final String VIDEO_UPLOAD_TOPIC = TOPIC_PREFIX + "video-upload"; + 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); - rocketMQTemplate.convertAndSend(VIDEO_UPLOAD_TOPIC, videoId); + log.info("准备发送视频上传消息: videoId={}", 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); - rocketMQTemplate.convertAndSend(VIDEO_PROCESS_TOPIC, videoId); + log.info("准备发送视频处理消息: videoId={}", 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); - rocketMQTemplate.convertAndSend(VIDEO_REVIEW_TOPIC, videoId); + log.info("准备发送视频审核消息: videoId={}", 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); + } } /** diff --git a/video-content/src/main/resources/application.yml b/video-content/src/main/resources/application.yml index 6e177d6..e5e9568 100644 --- a/video-content/src/main/resources/application.yml +++ b/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 diff --git a/video-gateway/src/main/java/com/nuaa/video/gateway/config/LoadBalancerConfig.java b/video-gateway/src/main/java/com/nuaa/video/gateway/config/LoadBalancerConfig.java new file mode 100644 index 0000000..880c36d --- /dev/null +++ b/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(); + } +} \ No newline at end of file diff --git a/video-gateway/src/main/resources/application.yml b/video-gateway/src/main/resources/application.yml index 66798c2..a6c34a3 100644 --- a/video-gateway/src/main/resources/application.yml +++ b/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}