Merge pull request #838 from zpaeng/main

feat:Websocket服务拆到单独服务，主后台保持单工通信

backend/open-isle.env.example

@@ -40,4 +40,10 @@ OPENAI_API_KEY=<你的openai-api-key>
 WEBPUSH_PUBLIC_KEY=<你的webpush-public-key>
 WEBPUSH_PRIVATE_KEY=<你的webpush-private-key>
 
+# === RabbitMQ ===
+RABBITMQ_HOST=<你的rabbitmq_host>
+RABBITMQ_PORT=<你的rabbitmq_port>
+RABBITMQ_USERNAME=<你的rabbitmq_username>
+RABBITMQ_PASSWORD=<你的rabbitmq_password>
+
 # LOG_LEVEL=DEBUG

backend/pom.xml

@@ -27,8 +27,8 @@
       <artifactId>spring-boot-starter-web</artifactId>
     </dependency>
     <dependency>
-      <groupId>org.springframework.boot</groupId>
+        <groupId>org.springframework.boot</groupId>
-      <artifactId>spring-boot-starter-websocket</artifactId>
+        <artifactId>spring-boot-starter-amqp</artifactId>
     </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>

backend/src/main/java/com/openisle/config/RabbitMQConfig.java

@@ -0,0 +1,204 @@
+package com.openisle.config;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.context.annotation.DependsOn;
+
+import jakarta.annotation.PostConstruct;
+import java.util.ArrayList;
+import java.util.List;
+
+@Configuration
+@RequiredArgsConstructor
+public class RabbitMQConfig {
+
+    public static final String EXCHANGE_NAME = "openisle-exchange";
+    // 保持向后兼容的常量
+    public static final String QUEUE_NAME = "notifications-queue";
+    public static final String ROUTING_KEY = "notifications.routingkey";
+    
+    // 硬编码为16以匹配ShardingStrategy中的十六进制分片逻辑
+    private final int queueCount = 16;
+    
+    @Value("${rabbitmq.queue.durable}")
+    private boolean queueDurable;
+
+    @PostConstruct
+    public void init() {
+        System.out.println("RabbitMQ配置初始化: 队列数量=" + queueCount + ", 持久化=" + queueDurable);
+    }
+
+    @Bean
+    public TopicExchange exchange() {
+        return new TopicExchange(EXCHANGE_NAME);
+    }
+
+    /**
+     * 创建所有分片队列, 使用十六进制后缀 (0-f)
+     */
+    @Bean
+    public List<Queue> shardedQueues() {
+        System.out.println("开始创建分片队列 Bean...");
+        
+        List<Queue> queues = new ArrayList<>();
+        for (int i = 0; i < queueCount; i++) {
+            String shardKey = Integer.toHexString(i);
+            String queueName = "notifications-queue-" + shardKey;
+            Queue queue = new Queue(queueName, queueDurable);
+            queues.add(queue);
+        }
+        
+        System.out.println("分片队列 Bean 创建完成，总数: " + queues.size());
+        return queues;
+    }
+
+    /**
+     * 创建所有分片绑定, 使用十六进制路由键 (notifications.shard.0 - notifications.shard.f)
+     */
+    @Bean
+    public List<Binding> shardedBindings(TopicExchange exchange, @Qualifier("shardedQueues") List<Queue> shardedQueues) {
+        System.out.println("开始创建分片绑定 Bean...");
+        List<Binding> bindings = new ArrayList<>();
+        if (shardedQueues != null) {
+            for (Queue queue : shardedQueues) {
+                String queueName = queue.getName();
+                String shardKey = queueName.substring("notifications-queue-".length());
+                String routingKey = "notifications.shard." + shardKey;
+                Binding binding = BindingBuilder.bind(queue).to(exchange).with(routingKey);
+                bindings.add(binding);
+            }
+        }
+        
+        System.out.println("分片绑定 Bean 创建完成，总数: " + bindings.size());
+        return bindings;
+    }
+
+    /**
+     * 保持向后兼容的单队列配置（可选）
+     */
+    @Bean
+    public Queue legacyQueue() {
+        return new Queue(QUEUE_NAME, queueDurable);
+    }
+
+    /**
+     * 保持向后兼容的单队列绑定（可选）
+     */
+    @Bean
+    public Binding legacyBinding(Queue legacyQueue, TopicExchange exchange) {
+        return BindingBuilder.bind(legacyQueue).to(exchange).with(ROUTING_KEY);
+    }
+
+    @Bean
+    public Jackson2JsonMessageConverter messageConverter() {
+        ObjectMapper objectMapper = new ObjectMapper();
+        objectMapper.registerModule(new com.fasterxml.jackson.datatype.jsr310.JavaTimeModule());
+        objectMapper.disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+        return new Jackson2JsonMessageConverter(objectMapper);
+    }
+
+    @Bean
+    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
+        return new RabbitAdmin(connectionFactory);
+    }
+
+    @Bean
+    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
+        RabbitTemplate template = new RabbitTemplate(connectionFactory);
+        template.setMessageConverter(messageConverter());
+        return template;
+    }
+
+    /**
+     * 使用 CommandLineRunner 确保在应用完全启动后声明队列到 RabbitMQ
+     * 这样可以确保 RabbitAdmin 和所有 Bean 都已正确初始化
+     */
+    @Bean
+    @DependsOn({"rabbitAdmin", "shardedQueues", "exchange"})
+    public CommandLineRunner queueDeclarationRunner(RabbitAdmin rabbitAdmin,
+                                                   @Qualifier("shardedQueues") List<Queue> shardedQueues,
+                                                   TopicExchange exchange,
+                                                   Queue legacyQueue,
+                                                   @Qualifier("shardedBindings") List<Binding> shardedBindings,
+                                                   Binding legacyBinding) {
+        return args -> {
+            System.out.println("=== 开始主动声明 RabbitMQ 组件 ===");
+            
+            try {
+                // 声明交换
+                rabbitAdmin.declareExchange(exchange);
+                
+                // 声明分片队列 - 检查存在性
+                System.out.println("开始检查并声明 " + shardedQueues.size() + " 个分片队列...");
+                int successCount = 0;
+                int skippedCount = 0;
+                
+                for (Queue queue : shardedQueues) {
+                    String queueName = queue.getName();
+                    try {
+                        // 使用 declareQueue 的返回值判断队列是否已存在
+                        // 如果队列已存在且配置匹配，declareQueue 会返回现有队列信息
+                        // 如果不匹配或不存在，会创建新队列
+                        rabbitAdmin.declareQueue(queue);
+                        successCount++;
+                    } catch (org.springframework.amqp.AmqpIOException e) {
+                        if (e.getMessage().contains("PRECONDITION_FAILED") && e.getMessage().contains("durable")) {
+                            skippedCount++;
+                        }
+                    } catch (Exception e) {
+                        System.err.println("队列声明失败: " + queueName + ", 错误: " + e.getMessage());
+                    }
+                }
+                System.out.println("分片队列处理完成: 成功 " + successCount + ", 跳过 " + skippedCount + ", 总数 " + shardedQueues.size());
+                
+                // 声明分片绑定
+                System.out.println("开始声明 " + shardedBindings.size() + " 个分片绑定...");
+                int bindingSuccessCount = 0;
+                for (Binding binding : shardedBindings) {
+                    try {
+                        rabbitAdmin.declareBinding(binding);
+                        bindingSuccessCount++;
+                    } catch (Exception e) {
+                        System.err.println("绑定声明失败: " + e.getMessage());
+                    }
+                }
+                System.out.println("分片绑定声明完成: 成功 " + bindingSuccessCount + "/" + shardedBindings.size());
+                
+                // 声明遗留队列和绑定 - 检查存在性
+                try {
+                    rabbitAdmin.declareQueue(legacyQueue);
+                    rabbitAdmin.declareBinding(legacyBinding);
+                    System.out.println("遗留队列和绑定就绪: " + QUEUE_NAME + " (已存在或新创建)");
+                } catch (org.springframework.amqp.AmqpIOException e) {
+                    if (e.getMessage().contains("PRECONDITION_FAILED") && e.getMessage().contains("durable")) {
+                        System.out.println("遗留队列已存在但 durable 设置不匹配: " + QUEUE_NAME + ", 保持现有队列");
+                    } else {
+                        System.err.println("遗留队列声明失败: " + QUEUE_NAME + ", 错误: " + e.getMessage());
+                    }
+                } catch (Exception e) {
+                    System.err.println("遗留队列声明失败: " + QUEUE_NAME + ", 错误: " + e.getMessage());
+                }
+                
+                System.out.println("=== RabbitMQ 组件声明完成 ===");
+                System.out.println("请检查 RabbitMQ 管理界面确认队列已正确创建");
+                
+            } catch (Exception e) {
+                System.err.println("RabbitMQ 组件声明过程中发生严重错误:");
+                e.printStackTrace();
+            }
+        };
+    }
+}

backend/src/main/java/com/openisle/config/SecurityConfig.java

@@ -74,10 +74,14 @@ public class SecurityConfig {
         CorsConfiguration cfg = new CorsConfiguration();
         cfg.setAllowedOrigins(List.of(
                 "http://127.0.0.1:8080",
+                "http://127.0.0.1:8081",
+                "http://127.0.0.1:8082",
                 "http://127.0.0.1:3000",
                 "http://127.0.0.1:3001",
                 "http://127.0.0.1",
                 "http://localhost:8080",
+                "http://localhost:8081",
+                "http://localhost:8082",
                 "http://localhost:3000",
                 "http://localhost:3001",
                 "http://localhost",

backend/src/main/java/com/openisle/config/ShardInfo.java

@@ -0,0 +1,14 @@
+package com.openisle.config;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class ShardInfo {
+    private int shardIndex;
+    private String queueName;
+    private String routingKey;
+}

backend/src/main/java/com/openisle/config/ShardingStrategy.java

@@ -0,0 +1,84 @@
+package com.openisle.config;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+@Component
+@Slf4j
+public class ShardingStrategy {
+    
+    // 固定为16以匹配RabbitMQConfig中的十六进制分片逻辑
+    private static final int QUEUE_COUNT = 16;
+    
+    // 分片分布统计
+    private final Map<Integer, AtomicLong> shardCounts = new ConcurrentHashMap<>();
+    
+    /**
+     * 根据用户名获取分片信息（基于哈希值首字符）
+     */
+    public ShardInfo getShardInfo(String username) {
+        if (username == null || username.isEmpty()) {
+            // 空用户名默认分到第0个分片
+            return getShardInfoByIndex(0);
+        }
+        
+        // 计算用户名的哈希值并转为十六进制字符串
+        String hash = Integer.toHexString(Math.abs(username.hashCode()));
+        
+        // 取哈希值的第一个字符 (0-9, a-f)
+        char firstChar = hash.charAt(0);
+        
+        // 十六进制字符映射到队列
+        int shard = getShardFromHexChar(firstChar);
+        recordShardUsage(shard);
+        
+        log.debug("Username '{}' -> hash '{}' -> firstChar '{}' -> shard {}", 
+                 username, hash, firstChar, shard);
+        
+        return getShardInfoByIndex(shard);
+    }
+    
+    /**
+     * 将十六进制字符映射到分片索引
+     */
+    private int getShardFromHexChar(char hexChar) {
+        int charValue;
+        if (hexChar >= '0' && hexChar <= '9') {
+            charValue = hexChar - '0'; // 0-9
+        } else if (hexChar >= 'a' && hexChar <= 'f') {
+            charValue = hexChar - 'a' + 10; // 10-15
+        } else {
+            // 异常情况，默认为0
+            charValue = 0;
+        }
+        
+        // 映射到队列数量范围内
+        return charValue % QUEUE_COUNT;
+    }
+    
+    /**
+     * 根据分片索引获取分片信息
+     */
+    private ShardInfo getShardInfoByIndex(int shard) {
+        String shardKey = Integer.toHexString(shard);
+        return new ShardInfo(
+            shard,
+            "notifications-queue-" + shardKey,
+            "notifications.shard." + shardKey
+        );
+    }
+    
+    /**
+     * 记录分片使用统计
+     */
+    private void recordShardUsage(int shard) {
+        shardCounts.computeIfAbsent(shard, k -> new AtomicLong(0)).incrementAndGet();
+    }
+
+}

backend/src/main/java/com/openisle/config/WebSocketConfig.java

@@ -1,110 +0,0 @@
-package com.openisle.config;
-
-import com.openisle.service.JwtService;
-import lombok.RequiredArgsConstructor;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageChannel;
-import org.springframework.messaging.simp.config.ChannelRegistration;
-import org.springframework.messaging.simp.config.MessageBrokerRegistry;
-import org.springframework.messaging.simp.stomp.StompCommand;
-import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
-import org.springframework.messaging.support.ChannelInterceptor;
-import org.springframework.messaging.support.MessageHeaderAccessor;
-import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
-import org.springframework.security.core.Authentication;
-import org.springframework.security.core.userdetails.UserDetailsService;
-import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
-import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
-import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
-
-@Configuration
-@EnableWebSocketMessageBroker
-@RequiredArgsConstructor
-public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
-    
-    private final JwtService jwtService;
-    private final UserDetailsService userDetailsService;
-    @Value("${app.website-url}")
-    private String websiteUrl;
-
-    @Override
-    public void configureMessageBroker(MessageBrokerRegistry config) {
-        // Enable a simple memory-based message broker to carry the messages back to the client on destinations prefixed with "/topic" and "/queue"
-        config.enableSimpleBroker("/topic", "/queue");
-        // Set user destination prefix for personal messages
-        config.setUserDestinationPrefix("/user");
-        // Designates the "/app" prefix for messages that are bound for @MessageMapping-annotated methods.
-        config.setApplicationDestinationPrefixes("/app");
-    }
-
-    @Override
-    public void registerStompEndpoints(StompEndpointRegistry registry) {
-        // 1) 原生 WebSocket（不带 SockJS）
-        registry.addEndpoint("/api/ws")
-                .setAllowedOriginPatterns(
-                        "https://staging.open-isle.com",
-                        "https://www.staging.open-isle.com",
-                        websiteUrl,
-                        websiteUrl.replace("://www.", "://"),
-                        "http://localhost:*",
-                        "http://127.0.0.1:*",
-                        "http://192.168.7.98:*",
-                        "http://30.211.97.238:*"
-                );
-
-        // 2) SockJS 回退：单独路径
-        registry.addEndpoint("/api/sockjs")
-                .setAllowedOriginPatterns(
-                        "https://staging.open-isle.com",
-                        "https://www.staging.open-isle.com",
-                        websiteUrl,
-                        websiteUrl.replace("://www.", "://"),
-                        "http://localhost:*",
-                        "http://127.0.0.1:*",
-                        "http://192.168.7.98:*",
-                        "http://30.211.97.238:*"
-                )
-                .withSockJS()
-                .setWebSocketEnabled(true)
-                .setSessionCookieNeeded(false);
-    }
-
-
-
-    @Override
-    public void configureClientInboundChannel(ChannelRegistration registration) {
-        registration.interceptors(new ChannelInterceptor() {
-            @Override
-            public Message<?> preSend(Message<?> message, MessageChannel channel) {
-                StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
-                
-                if (StompCommand.CONNECT.equals(accessor.getCommand())) {
-                    System.out.println("WebSocket CONNECT command received");
-                    String authHeader = accessor.getFirstNativeHeader("Authorization");
-                    System.out.println("Authorization header: " + (authHeader != null ? "present" : "missing"));
-                    
-                    if (authHeader != null && authHeader.startsWith("Bearer ")) {
-                        String token = authHeader.substring(7);
-                        try {
-                            String username = jwtService.validateAndGetSubject(token);
-                            System.out.println("JWT validated for user: " + username);
-                            var userDetails = userDetailsService.loadUserByUsername(username);
-                            Authentication auth = new UsernamePasswordAuthenticationToken(
-                                userDetails, null, userDetails.getAuthorities());
-                            accessor.setUser(auth);
-                            System.out.println("WebSocket user set: " + username);
-                        } catch (Exception e) {
-                            System.err.println("JWT validation failed: " + e.getMessage());
-                        }
-                    }
-                } else if (StompCommand.SUBSCRIBE.equals(accessor.getCommand())) {
-                    System.out.println("WebSocket SUBSCRIBE to: " + accessor.getDestination());
-                    System.out.println("WebSocket user during subscribe: " + (accessor.getUser() != null ? accessor.getUser().getName() : "null"));
-                }
-                return message;
-            }
-        });
-    }
-}

backend/src/main/java/com/openisle/dto/MessageNotificationPayload.java

@@ -0,0 +1,15 @@
+package com.openisle.dto;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.io.Serializable;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class MessageNotificationPayload implements Serializable {
+    private String targetUsername;
+    private Object payload;
+}

backend/src/main/java/com/openisle/model/Message.java

@@ -1,5 +1,6 @@
 package com.openisle.model;
 
+import com.fasterxml.jackson.annotation.JsonBackReference;
 import jakarta.persistence.*;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -20,6 +21,7 @@ public class Message {
 
     @ManyToOne(optional = false, fetch = FetchType.LAZY)
     @JoinColumn(name = "conversation_id")
+    @JsonBackReference
     private MessageConversation conversation;
 
     @ManyToOne(optional = false, fetch = FetchType.LAZY)

backend/src/main/java/com/openisle/model/MessageConversation.java

@@ -1,5 +1,7 @@
 package com.openisle.model;
 
+import com.fasterxml.jackson.annotation.JsonBackReference;
+import com.fasterxml.jackson.annotation.JsonManagedReference;
 import jakarta.persistence.*;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -41,8 +43,10 @@ public class MessageConversation {
     private Message lastMessage;
 
     @OneToMany(mappedBy = "conversation", cascade = CascadeType.ALL, orphanRemoval = true)
+    @JsonBackReference
     private Set<MessageParticipant> participants = new HashSet<>();
 
     @OneToMany(mappedBy = "conversation", cascade = CascadeType.ALL, orphanRemoval = true)
+    @JsonBackReference
     private Set<Message> messages = new HashSet<>();
 }

backend/src/main/java/com/openisle/model/MessageParticipant.java

@@ -1,5 +1,6 @@
 package com.openisle.model;
 
+import com.fasterxml.jackson.annotation.JsonBackReference;
 import jakarta.persistence.*;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -19,6 +20,7 @@ public class MessageParticipant {
 
     @ManyToOne(optional = false, fetch = FetchType.LAZY)
     @JoinColumn(name = "conversation_id")
+    @JsonBackReference
     private MessageConversation conversation;
 
     @ManyToOne(optional = false, fetch = FetchType.LAZY)

backend/src/main/java/com/openisle/repository/MessageConversationRepository.java

@@ -11,6 +11,9 @@ import java.util.List;
 
 @Repository
 public interface MessageConversationRepository extends JpaRepository<MessageConversation, Long> {
+
+    @Query("SELECT c FROM MessageConversation c LEFT JOIN FETCH c.participants p LEFT JOIN FETCH p.user WHERE c.id = :id")
+    java.util.Optional<MessageConversation> findByIdWithParticipantsAndUsers(@Param("id") Long id);
     @Query("SELECT c FROM MessageConversation c " +
            "WHERE c.channel = false AND size(c.participants) = 2 " +
            "AND EXISTS (SELECT 1 FROM c.participants p1 WHERE p1.user = :user1) " +

backend/src/main/java/com/openisle/service/MessageService.java

@@ -16,16 +16,18 @@ import com.openisle.dto.MessageDto;
 import com.openisle.dto.ReactionDto;
 import com.openisle.dto.UserSummaryDto;
 import com.openisle.mapper.ReactionMapper;
+import com.openisle.dto.MessageNotificationPayload;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
-import org.springframework.messaging.simp.SimpMessagingTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.time.LocalDateTime;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 @Service
@@ -37,7 +39,7 @@ public class MessageService {
     private final MessageConversationRepository conversationRepository;
     private final MessageParticipantRepository participantRepository;
     private final UserRepository userRepository;
-    private final SimpMessagingTemplate messagingTemplate;
+    private final NotificationProducer notificationProducer;
     private final ReactionRepository reactionRepository;
     private final ReactionMapper reactionMapper;
 
@@ -69,26 +71,41 @@ public class MessageService {
         conversationRepository.save(conversation);
         log.info("Conversation {} updated with last message ID {}", conversation.getId(), message.getId());
 
-        // Broadcast the new message to subscribed clients
-        MessageDto messageDto = toDto(message);
-        String conversationDestination = "/topic/conversation/" + conversation.getId();
-        messagingTemplate.convertAndSend(conversationDestination, messageDto);
-        log.info("Message {} broadcasted to destination: {}", message.getId(), conversationDestination);
-
-        // Also notify the recipient on their personal channel to update the conversation list
-        String userDestination = "/topic/user/" + recipient.getId() + "/messages";
-        messagingTemplate.convertAndSend(userDestination, messageDto);
-        log.info("Message {} notification sent to destination: {}", message.getId(), userDestination);
-
-        // Notify recipient of new unread count
-        long unreadCount = getUnreadMessageCount(recipientId);
-        log.info("Calculating unread count for user {}: {}", recipientId, unreadCount);
         
-        // Send using username instead of user ID for WebSocket routing
+        try {
-        String recipientUsername = recipient.getUsername();
+            MessageDto messageDto = toDto(message);
-        messagingTemplate.convertAndSendToUser(recipientUsername, "/queue/unread-count", unreadCount);
+            
-        log.info("Sent unread count {} to user {} (username: {}) via WebSocket destination: /user/{}/queue/unread-count",
+            long unreadCount = getUnreadMessageCount(recipientId);
-                unreadCount, recipientId, recipientUsername, recipientUsername);
+
+            // 创建包含对话和参与者信息的完整payload
+            Map<String, Object> conversationInfo = new HashMap<>();
+            conversationInfo.put("id", conversation.getId());
+            conversationInfo.put("participants", conversation.getParticipants().stream()
+                .map(p -> {
+                    Map<String, Object> participantInfo = new HashMap<>();
+                    participantInfo.put("userId", p.getUser().getId());
+                    participantInfo.put("username", p.getUser().getUsername());
+                    return participantInfo;
+                }).collect(Collectors.toList()));
+            
+            Map<String, Object> combinedPayload = new HashMap<>();
+            combinedPayload.put("message", messageDto);
+            combinedPayload.put("unreadCount", unreadCount);
+            combinedPayload.put("conversation", conversationInfo);
+            combinedPayload.put("senderId", senderId);
+            if (notificationProducer != null) {
+                log.info("NotificationProducer is available");
+            } else {
+                log.info("ERROR: NotificationProducer is NULL!");
+                return message;
+            }
+            log.info("Recipient username: {}", recipient.getUsername());
+            
+            notificationProducer.sendNotification(new MessageNotificationPayload(recipient.getUsername(), combinedPayload));
+            log.info("=== Notification call completed ===");
+        } catch (Exception e) {
+            log.error("=== Error in notification process ===", e);
+        }
 
         return message;
     }
@@ -97,7 +114,7 @@ public class MessageService {
     public Message sendMessageToConversation(Long senderId, Long conversationId, String content, Long replyToId) {
         User sender = userRepository.findById(senderId)
                 .orElseThrow(() -> new IllegalArgumentException("Sender not found"));
-        MessageConversation conversation = conversationRepository.findById(conversationId)
+        MessageConversation conversation = conversationRepository.findByIdWithParticipantsAndUsers(conversationId)
                 .orElseThrow(() -> new IllegalArgumentException("Conversation not found"));
 
         // Join the conversation if not already a participant (useful for channels)
@@ -124,22 +141,30 @@ public class MessageService {
         conversationRepository.save(conversation);
 
         MessageDto messageDto = toDto(message);
-        String conversationDestination = "/topic/conversation/" + conversation.getId();
-        messagingTemplate.convertAndSend(conversationDestination, messageDto);
 
-        // Notify all participants except sender for updates
+        // Build participant payloads once to avoid duplicate broadcasts
-        for (MessageParticipant participant : conversation.getParticipants()) {
+        java.util.List<Map<String, Object>> participantInfos = conversation.getParticipants().stream()
-            if (participant.getUser().getId().equals(senderId)) continue;
+                .filter(p -> !p.getUser().getId().equals(senderId))
-            String userDestination = "/topic/user/" + participant.getUser().getId() + "/messages";
+                .map(p -> {
-            messagingTemplate.convertAndSend(userDestination, messageDto);
+                    Map<String, Object> info = new HashMap<>();
+                    info.put("userId", p.getUser().getId());
+                    info.put("username", p.getUser().getUsername());
+                    info.put("unreadCount", getUnreadMessageCount(p.getUser().getId()));
+                    info.put("channelUnread", getUnreadChannelCount(p.getUser().getId()));
+                    return info;
+                }).collect(Collectors.toList());
 
-            long unreadCount = getUnreadMessageCount(participant.getUser().getId());
+        Map<String, Object> conversationInfo = new HashMap<>();
-            String username = participant.getUser().getUsername();
+        conversationInfo.put("id", conversation.getId());
-            messagingTemplate.convertAndSendToUser(username, "/queue/unread-count", unreadCount);
+        conversationInfo.put("participants", participantInfos);
 
-            long channelUnread = getUnreadChannelCount(participant.getUser().getId());
+        Map<String, Object> combinedPayload = new HashMap<>();
-            messagingTemplate.convertAndSendToUser(username, "/queue/channel-unread", channelUnread);
+        combinedPayload.put("message", messageDto);
-        }
+        combinedPayload.put("conversation", conversationInfo);
+        combinedPayload.put("senderId", senderId);
+
+        // Use sender's username for sharding; only one notification is needed
+        notificationProducer.sendNotification(new MessageNotificationPayload(sender.getUsername(), combinedPayload));
 
         return message;
     }

backend/src/main/java/com/openisle/service/NotificationProducer.java

@@ -0,0 +1,63 @@
+package com.openisle.service;
+
+import com.openisle.config.RabbitMQConfig;
+import com.openisle.config.ShardInfo;
+import com.openisle.config.ShardingStrategy;
+import com.openisle.dto.MessageNotificationPayload;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class NotificationProducer {
+
+    private final RabbitTemplate rabbitTemplate;
+    private final ShardingStrategy shardingStrategy;
+    
+    @Value("${rabbitmq.sharding.enabled}")
+    private boolean shardingEnabled;
+
+    public void sendNotification(MessageNotificationPayload payload) {
+        String targetUsername = payload.getTargetUsername();
+        
+        try {
+            if (shardingEnabled) {
+                // 使用分片策略发送消息
+                sendShardedNotification(payload, targetUsername);
+            } else {
+                // 使用原始单队列方式发送（向后兼容）
+                sendLegacyNotification(payload);
+            }
+        } catch (Exception e) {
+            log.error("Failed to send message to RabbitMQ for user: {}", targetUsername, e);
+            throw e;
+        }
+    }
+    
+    /**
+     * 使用分片策略发送消息
+     */
+    private void sendShardedNotification(MessageNotificationPayload payload, String targetUsername) {
+        ShardInfo shardInfo = shardingStrategy.getShardInfo(targetUsername);
+        rabbitTemplate.convertAndSend(
+            RabbitMQConfig.EXCHANGE_NAME,
+            shardInfo.getRoutingKey(),
+            payload
+        );
+    }
+    
+    /**
+     * 使用原始单队列方式发送消息（向后兼容）
+     */
+    private void sendLegacyNotification(MessageNotificationPayload payload) {
+        rabbitTemplate.convertAndSend(
+            RabbitMQConfig.EXCHANGE_NAME,
+            RabbitMQConfig.ROUTING_KEY,
+            payload
+        );
+    }
+}

backend/src/main/resources/application.properties

@@ -87,6 +87,16 @@ app.website-url=${WEBSITE_URL:https://www.open-isle.com}
 app.webpush.public-key=${WEBPUSH_PUBLIC_KEY:}
 app.webpush.private-key=${WEBPUSH_PRIVATE_KEY:}
 
+# RabbitMQ Configuration
+spring.rabbitmq.host=${RABBITMQ_HOST:localhost}
+spring.rabbitmq.port=${RABBITMQ_PORT:5672}
+spring.rabbitmq.username=${RABBITMQ_USERNAME:guest}
+spring.rabbitmq.password=${RABBITMQ_PASSWORD:guest}
+
+# RabbitMQ 队列配置 - 修改为非持久化以匹配现有队列
+rabbitmq.queue.durable=true
+rabbitmq.sharding.enabled=true
+
 # springdoc-openapi-starter-webmvc-api
 # see https://springdoc.org/#springdoc-openapi-core-properties
 springdoc.api-docs.path=/v3/api-docs

frontend_nuxt/.env.dev.example

@@ -1,5 +1,6 @@
 ; 本地部署后端
 NUXT_PUBLIC_API_BASE_URL=https://127.0.0.1:8081
+NUXT_PUBLIC_WEBSOCKET_URL=https://127.0.0.1:8082
 NUXT_PUBLIC_WEBSITE_BASE_URL=http://localhost:3000
 
 NUXT_PUBLIC_GOOGLE_CLIENT_ID=777830451304-nt8afkkap18gui4f9entcha99unal744.apps.googleusercontent.com

frontend_nuxt/.env.example

@@ -5,6 +5,9 @@
 ; 生产环境后端
 NUXT_PUBLIC_API_BASE_URL=https://open-isle.com
 
+; 生产环境ws后端
+NUXT_PUBLIC_WEBSOCKET_URL=https://open-isle.com/websocket
+
 ; 预发环境
 ; NUXT_PUBLIC_WEBSITE_BASE_URL=https://staging.open-isle.com
 ; 正式环境/生产环境

frontend_nuxt/.env.production.example

@@ -3,6 +3,8 @@
 NUXT_PUBLIC_API_BASE_URL=https://open-isle.com
 ; 正式环境/生产环境
 NUXT_PUBLIC_WEBSITE_BASE_URL=https://open-isle.com
+; 生产环境ws后端
+NUXT_PUBLIC_WEBSOCKET_URL=https://open-isle.com/websocket
 
 NUXT_PUBLIC_GOOGLE_CLIENT_ID=777830451304-nt8afkkap18gui4f9entcha99unal744.apps.googleusercontent.com
 NUXT_PUBLIC_GITHUB_CLIENT_ID=Ov23liVkO1NPAX5JyWxJ

frontend_nuxt/.env.staging.example

@@ -1,5 +1,12 @@
+; 本地部署后端
+; NUXT_PUBLIC_API_BASE_URL=https://127.0.0.1:8081
+
 ; 预发环境后端
 NUXT_PUBLIC_API_BASE_URL=https://staging.open-isle.com
+
+; 预发环境ws后端
+NUXT_PUBLIC_WEBSOCKET_URL=https://staging.open-isle.com/websocket
+
 ; 预发环境
 NUXT_PUBLIC_WEBSITE_BASE_URL=https://staging.open-isle.com
 

frontend_nuxt/composables/useChannelsUnreadCount.js

@@ -3,82 +3,73 @@ import { useWebSocket } from './useWebSocket'
 import { getToken } from '~/utils/auth'
 
 const count = ref(0)
-let isInitialized = false
+let isInitialized = false;
-let wsSubscription = null
 
 export function useChannelsUnreadCount() {
-  const config = useRuntimeConfig()
+  const config = useRuntimeConfig();
-  const API_BASE_URL = config.public.apiBaseUrl
+  const API_BASE_URL = config.public.apiBaseUrl;
-  const { subscribe, isConnected, connect } = useWebSocket()
+  const { subscribe, isConnected, connect } = useWebSocket();
 
   const fetchChannelUnread = async () => {
-    const token = getToken()
+    const token = getToken();
     if (!token) {
-      count.value = 0
+      count.value = 0;
-      return
+      return;
     }
     try {
       const response = await fetch(`${API_BASE_URL}/api/channels/unread-count`, {
         headers: { Authorization: `Bearer ${token}` },
-      })
+      });
       if (response.ok) {
-        const data = await response.json()
+        const data = await response.json();
-        count.value = data
+        count.value = data;
       }
     } catch (e) {
-      console.error('Failed to fetch channel unread count:', e)
+      console.error('Failed to fetch channel unread count:', e);
     }
-  }
+  };
+  
+  const setupWebSocketListener = () => {
+    const destination = '/user/queue/channel-unread';
+
+    subscribe(destination, (message) => {
+      const unread = parseInt(message.body, 10);
+      if (!isNaN(unread)) {
+        count.value = unread;
+      }
+    }).then(subscription => {
+      if (subscription) {
+        console.log('频道未读消息订阅成功');
+      }
+    });
+  };
 
   const initialize = () => {
-    const token = getToken()
+    const token = getToken();
     if (!token) {
-      count.value = 0
+      count.value = 0;
-      return
+      return;
     }
-    fetchChannelUnread()
-    if (!isConnected.value) {
-      connect(token)
-    }
-    setupWebSocketListener()
-  }
 
-  const setupWebSocketListener = () => {
+    if (!isConnected.value) {
-    if (!wsSubscription) {
+      connect(token);
-      watch(
-        isConnected,
-        (newValue) => {
-          if (newValue && !wsSubscription) {
-            wsSubscription = subscribe('/user/queue/channel-unread', (message) => {
-              const unread = parseInt(message.body, 10)
-              if (!isNaN(unread)) {
-                count.value = unread
-              }
-            })
-          }
-        },
-        { immediate: true },
-      )
     }
-  }
+
+    fetchChannelUnread();
+    setupWebSocketListener();
+  };
 
   const setFromList = (channels) => {
-    count.value = Array.isArray(channels) ? channels.filter((c) => c.unreadCount > 0).length : 0
+    count.value = Array.isArray(channels) ? channels.filter((c) => c.unreadCount > 0).length : 0;
-  }
+  };
 
-  const hasUnread = computed(() => count.value > 0)
+  const hasUnread = computed(() => count.value > 0);
 
-  const token = getToken()
+  if (!isInitialized) {
-  if (token) {
+    const token = getToken();
-    if (!isInitialized) {
+    if (token) {
-      isInitialized = true
+      isInitialized = true;
-      initialize()
+      initialize();
-    } else {
-      fetchChannelUnread()
-      if (!isConnected.value) {
-        connect(token)
-      }
-      setupWebSocketListener()
     }
   }
 
@@ -88,5 +79,5 @@ export function useChannelsUnreadCount() {
     fetchChannelUnread,
     initialize,
     setFromList,
-  }
+  };
 }

frontend_nuxt/composables/useUnreadCount.js

@@ -4,7 +4,6 @@ import { getToken } from '~/utils/auth';
 
 const count = ref(0);
 let isInitialized = false;
-let wsSubscription = null;
 
 export function useUnreadCount() {
   const config = useRuntimeConfig();
@@ -30,64 +29,48 @@ export function useUnreadCount() {
     }
   };
 
-  const initialize = async () => {
+  const setupWebSocketListener = () => {
+    console.log('设置未读消息订阅...');
+    const destination = '/user/queue/unread-count';
+    
+    subscribe(destination, (message) => {
+      const unreadCount = parseInt(message.body, 10);
+      if (!isNaN(unreadCount)) {
+        count.value = unreadCount;
+      }
+    }).then(subscription => {
+      if (subscription) {
+        console.log('未读消息订阅成功');
+      }
+    });
+  };
+
+  const initialize = () => {
     const token = getToken();
     if (!token) {
       count.value = 0;
       return;
     }
 
-    // 总是获取最新的未读数量
-    fetchUnreadCount();
-    
-    // 确保WebSocket连接
     if (!isConnected.value) {
       connect(token);
     }
     
-    // 设置WebSocket监听
+    fetchUnreadCount();
-    await setupWebSocketListener();
+    setupWebSocketListener();
   };
 
-  const setupWebSocketListener = async () => {
+  if (!isInitialized) {
-    // 只有在还没有订阅的情况下才设置监听
+    const token = getToken();
-    if (!wsSubscription) {
+    if (token) {
-      
-      watch(isConnected, (newValue) => {
-        if (newValue && !wsSubscription) {
-          const destination = `/user/queue/unread-count`;
-          wsSubscription = subscribe(destination, (message) => {
-            const unreadCount = parseInt(message.body, 10);
-            if (!isNaN(unreadCount)) {
-              count.value = unreadCount;
-            }
-          });
-        }
-      }, { immediate: true });
-    }
-  };
-
-  // 自动初始化逻辑 - 确保每次调用都能获取到未读数量并设置监听
-  const token = getToken();
-  if (token) {
-    if (!isInitialized) {
       isInitialized = true;
-      initialize(); // 完整初始化，包括WebSocket监听
+      initialize();
-    } else {
-      // 即使已经初始化，也要确保获取最新的未读数量并确保WebSocket监听存在
-      fetchUnreadCount();
-      
-      // 确保WebSocket连接和监听都存在
-      if (!isConnected.value) {
-        connect(token);
-      }
-      setupWebSocketListener();
     }
   }
 
   return {
     count,
     fetchUnreadCount,
-    initialize,
+    initialize, 
   };
 }

frontend_nuxt/composables/useWebSocket.js

@@ -1,86 +1,182 @@
-import { ref } from 'vue'
+import { ref, readonly, watch } from 'vue'
 import { Client } from '@stomp/stompjs'
 import SockJS from 'sockjs-client/dist/sockjs.min.js'
 import { useRuntimeConfig } from '#app'
 
 const client = ref(null)
 const isConnected = ref(false)
+const activeSubscriptions = ref(new Map())
+// Store callbacks to allow for re-subscription after reconnect
+const resubscribeCallbacks = new Map()
+
+// Helper for unified subscription logging
+const logSubscriptionActivity = (action, destination, subscriptionId = 'N/A') => {
+  console.log(
+    `[SUB_MAN] ${action} | Dest: ${destination} | SubID: ${subscriptionId} | Active: ${activeSubscriptions.value.size}`
+  )
+}
 
 const connect = (token) => {
-  if (isConnected.value) {
+  if (isConnected.value || (client.value && client.value.active)) {
     return
   }
 
-  const config = useRuntimeConfig()
-  const API_BASE_URL = config.public.apiBaseUrl
-  const socketUrl = `${API_BASE_URL}/api/sockjs`
 
-  const socket = new SockJS(socketUrl)
+  const config = useRuntimeConfig()
+  const WEBSOCKET_URL = config.public.websocketUrl
+  const socketUrl = `${WEBSOCKET_URL}/api/sockjs`
+
   const stompClient = new Client({
-    webSocketFactory: () => socket,
+    webSocketFactory: () => new SockJS(socketUrl),
     connectHeaders: {
       Authorization: `Bearer ${token}`,
     },
-    debug: function (str) {},
+    debug: function (str) {
-    reconnectDelay: 5000,
+    
+    },
+    reconnectDelay: 10000,
     heartbeatIncoming: 4000,
     heartbeatOutgoing: 4000,
   })
 
   stompClient.onConnect = (frame) => {
     isConnected.value = true
+    resubscribeCallbacks.forEach((callback, destination) => {
+        doSubscribe(destination, callback)
+    })
   }
 
   stompClient.onStompError = (frame) => {
-    console.error('WebSocket STOMP error:', frame)
+    console.error('Full frame:', frame)
+  }
+
+  stompClient.onWebSocketError = (event) => {
+    
+  }
+
+  stompClient.onWebSocketClose = (event) => {
+    isConnected.value = false;
+    activeSubscriptions.value.clear();
+    logSubscriptionActivity('Cleared all subscriptions due to WebSocket close', 'N/A');
+  };
+  
+  stompClient.onDisconnect = (frame) => {
+    isConnected.value = false
   }
 
   stompClient.activate()
   client.value = stompClient
 }
 
+const unsubscribe = (destination) => {
+  if (!destination) {
+    return false
+  }
+  const subscription = activeSubscriptions.value.get(destination)
+  if (subscription) {
+    try {
+      subscription.unsubscribe()
+      logSubscriptionActivity('Unsubscribed', destination, subscription.id)
+    } catch (e) {
+      console.error(`Error during unsubscribe for ${destination}:`, e)
+    } finally {
+      activeSubscriptions.value.delete(destination)
+      resubscribeCallbacks.delete(destination)
+    }
+    return true
+  } else {
+    return false
+  }
+}
+
+const unsubscribeAll = () => {
+  logSubscriptionActivity('Unsubscribing from ALL', `Total: ${activeSubscriptions.value.size}`)
+  const destinations = [...activeSubscriptions.value.keys()]
+  destinations.forEach(dest => {
+    unsubscribe(dest)
+  })
+}
+
 const disconnect = () => {
+  unsubscribeAll()
   if (client.value) {
-    isConnected.value = false
+    try {
-    client.value.deactivate()
+      client.value.deactivate()
+    } catch (e) {
+      console.error('Error during client deactivation:', e)
+    }
     client.value = null
+    isConnected.value = false
+  }
+}
+
+const doSubscribe = (destination, callback) => {
+  try {
+    if (!client.value || !client.value.connected) {
+      return null
+    }
+
+    if (activeSubscriptions.value.has(destination)) {
+      unsubscribe(destination)
+    }
+
+    const subscription = client.value.subscribe(destination, (message) => {
+      callback(message)
+    })
+
+    if (subscription) {
+      activeSubscriptions.value.set(destination, subscription)
+      resubscribeCallbacks.set(destination, callback) // Store for re-subscription
+      logSubscriptionActivity('Subscribed', destination, subscription.id)
+      return subscription
+    } else {
+      return null
+    }
+  } catch (error) {
+    console.error(`Exception during subscription to ${destination}:`, error)
+    return null
   }
 }
 
 const subscribe = (destination, callback) => {
-  if (!isConnected.value || !client.value || !client.value.connected) {
+  if (!destination) {
-    return null
+    return Promise.resolve(null)
   }
 
-  try {
+  return new Promise((resolve) => {
-    const subscription = client.value.subscribe(destination, (message) => {
+    if (client.value && client.value.connected) {
-      try {
+      const sub = doSubscribe(destination, callback)
-        if (
+      resolve(sub)
-          destination.includes('/queue/unread-count') ||
+    } else {
-          destination.includes('/queue/channel-unread')
+      const unwatch = watch(isConnected, (newVal) => {
-        ) {
+        if (newVal) {
-          callback(message)
+          setTimeout(() => {
-        } else {
+            const sub = doSubscribe(destination, callback)
-          const parsedMessage = JSON.parse(message.body)
+            unwatch()
-          callback(parsedMessage)
+            resolve(sub)
+          }, 100)
         }
-      } catch (error) {
+      }, { immediate: false })
-        callback(message)
+      
-      }
+      setTimeout(() => {
-    })
+        unwatch()
-
+        if (!isConnected.value) {
-    return subscription
+          resolve(null)
-  } catch (error) {
+        }
-    return null
+      }, 15000)
-  }
+    }
+  })
 }
 
 export function useWebSocket() {
   return {
-    client,
+    client: readonly(client),
     isConnected,
     connect,
     disconnect,
     subscribe,
+    unsubscribe,
+    unsubscribeAll,
+    activeSubscriptions: readonly(activeSubscriptions),
   }
 }

frontend_nuxt/nuxt.config.ts

@@ -6,6 +6,7 @@ export default defineNuxtConfig({
   runtimeConfig: {
     public: {
       apiBaseUrl: process.env.NUXT_PUBLIC_API_BASE_URL || '',
+      websocketUrl: process.env.NUXT_PUBLIC_WEBSOCKET_URL || '',
       websiteBaseUrl: process.env.NUXT_PUBLIC_WEBSITE_BASE_URL || '',
       googleClientId: process.env.NUXT_PUBLIC_GOOGLE_CLIENT_ID || '',
       githubClientId: process.env.NUXT_PUBLIC_GITHUB_CLIENT_ID || '',

frontend_nuxt/pages/message-box/[id].vue

@@ -100,10 +100,9 @@ import BasePlaceholder from '~/components/BasePlaceholder.vue'
 const config = useRuntimeConfig()
 const route = useRoute()
 const API_BASE_URL = config.public.apiBaseUrl
-const { connect, disconnect, subscribe, isConnected } = useWebSocket()
+const { connect, subscribe, unsubscribe, isConnected } = useWebSocket()
 const { fetchUnreadCount: refreshGlobalUnreadCount } = useUnreadCount()
 const { fetchChannelUnread: refreshChannelUnread } = useChannelsUnreadCount()
-let subscription = null
 
 const messages = ref([])
 const participants = ref([])
@@ -338,8 +337,12 @@ onMounted(async () => {
     // 初次进入频道时，平滑滚动到底部
     scrollToBottomSmooth()
     const token = getToken()
-    if (token && !isConnected.value) {
+    if (token) {
-      connect(token)
+      if (isConnected.value) {
+        subscribeToConversation()
+      } else {
+        connect(token)
+      }
     }
   } else {
     toast.error('请先登录')
@@ -347,26 +350,39 @@ onMounted(async () => {
   }
 })
 
+const subscribeToConversation = () => {
+  if (!currentUser.value) return;
+  const destination = `/topic/conversation/${conversationId}`
+  
+  subscribe(destination, async (message) => {
+    try {
+      const parsedMessage = JSON.parse(message.body)
+
+      if (parsedMessage.sender && parsedMessage.sender.id === currentUser.value.id) {
+        return
+      }
+
+      messages.value.push({
+        ...parsedMessage,
+        src: parsedMessage.sender.avatar,
+        iconClick: () => openUser(parsedMessage.sender.id),
+      })
+
+      await markConversationAsRead()
+      await nextTick()
+      
+      if (isUserNearBottom.value) {
+        scrollToBottomSmooth()
+      }
+    } catch (e) {
+      console.error("Failed to parse websocket message", e)
+    }
+  })
+}
+
 watch(isConnected, (newValue) => {
   if (newValue) {
-    setTimeout(() => {
+    subscribeToConversation()
-      subscription = subscribe(`/topic/conversation/${conversationId}`, async (message) => {
-        // 避免重复显示当前用户发送的消息
-        if (message.sender.id !== currentUser.value.id) {
-          messages.value.push({
-            ...message,
-            src: message.sender.avatar,
-            iconClick: () => {
-              openUser(message.sender.id)
-            },
-          })
-          // 收到消息后只标记已读，不强制滚动（符合“非发送不拉底”）
-          markConversationAsRead()
-          await nextTick()
-          updateNearBottom()
-        }
-      })
-    }, 500)
   }
 })
 
@@ -378,7 +394,12 @@ onActivated(async () => {
     await nextTick()
     scrollToBottomSmooth()
     updateNearBottom()
-    if (!isConnected.value) {
+    
+    if (isConnected.value) {
+      // 如果已连接，重新订阅
+      subscribeToConversation()
+    } else {
+      // 如果未连接，则发起连接
       const token = getToken()
       if (token) connect(token)
     }
@@ -386,22 +407,17 @@ onActivated(async () => {
 })
 
 onDeactivated(() => {
-  if (subscription) {
+  const destination = `/topic/conversation/${conversationId}`
-    subscription.unsubscribe()
+  unsubscribe(destination)
-    subscription = null
-  }
-  disconnect()
 })
 
 onUnmounted(() => {
-  if (subscription) {
+  const destination = `/topic/conversation/${conversationId}`
-    subscription.unsubscribe()
+  unsubscribe(destination)
-    subscription = null
+
-  }
   if (messagesListEl.value) {
     messagesListEl.value.removeEventListener('scroll', updateNearBottom)
   }
-  disconnect()
 })
 
 function minimize() {

frontend_nuxt/pages/message-box/index.vue

@@ -118,7 +118,7 @@
 </template>
 
 <script setup>
-import { ref, onUnmounted, watch, onActivated, computed } from 'vue'
+import { ref, onUnmounted, watch, onActivated, computed, onDeactivated } from 'vue'
 import { useRoute } from 'vue-router'
 import { getToken, fetchCurrentUser } from '~/utils/auth'
 import { toast } from '~/main'
@@ -139,11 +139,10 @@ const error = ref(null)
 const route = useRoute()
 const currentUser = ref(null)
 const API_BASE_URL = config.public.apiBaseUrl
-const { connect, disconnect, subscribe, isConnected } = useWebSocket()
+const { connect, subscribe, unsubscribe, isConnected } = useWebSocket()
 const { fetchUnreadCount: refreshGlobalUnreadCount } = useUnreadCount()
 const { fetchChannelUnread: refreshChannelUnread, setFromList: setChannelUnreadFromList } =
   useChannelsUnreadCount()
-let subscription = null
 
 const activeTab = ref('channels')
 const tabs = [
@@ -259,37 +258,45 @@ onActivated(async () => {
     refreshGlobalUnreadCount()
     refreshChannelUnread()
     const token = getToken()
-    if (token && !isConnected.value) {
+    if (token) {
-      connect(token)
+      if (isConnected.value) {
+        // 如果已经连接，但可能因为组件销毁而取消了订阅，所以需要重新订阅
+        subscribeToUserMessages()
+      } else {
+        // 如果未连接，则发起连接，连接成功后 watch 回调会处理订阅
+        connect(token)
+      }
     }
   } else {
     loading.value = false
   }
 })
 
-watch(isConnected, (newValue) => {
+const subscribeToUserMessages = () => {
-  if (newValue && currentUser.value) {
+  if (!currentUser.value) return;
-    const destination = `/topic/user/${currentUser.value.id}/messages`
+  const destination = `/topic/user/${currentUser.value.id}/messages`
-
+  
-    // 清理旧的订阅
+  subscribe(destination, (message) => {
-    if (subscription) {
+    if (activeTab.value === 'messages') {
-      subscription.unsubscribe()
-    }
-
-    subscription = subscribe(destination, (message) => {
       fetchConversations()
-      if (activeTab.value === 'channels') {
+    }
-        fetchChannels()
+    fetchChannels()
-      }
+    refreshGlobalUnreadCount()
-    })
+    refreshChannelUnread()
+  })
+}
+
+watch(isConnected, (newValue) => {
+  if (newValue) {
+    subscribeToUserMessages()
   }
 })
 
-onUnmounted(() => {
+onDeactivated(() => {
-  if (subscription) {
+  if (currentUser.value) {
-    subscription.unsubscribe()
+    const destination = `/topic/user/${currentUser.value.id}/messages`
+    unsubscribe(destination)
   }
-  disconnect()
 })
 
 function goToConversation(id) {

websocket_service/pom.xml

@@ -0,0 +1,74 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>3.1.1</version>
+        <relativePath/> <!-- lookup parent from repository -->
+    </parent>
+    <groupId>com.openisle</groupId>
+    <artifactId>websocket-service</artifactId>
+    <version>0.0.1-SNAPSHOT</version>
+    <name>websocket-service</name>
+    <description>Dedicated WebSocket service for OpenIsle</description>
+    <properties>
+        <java.version>17</java.version>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-security</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.jsonwebtoken</groupId>
+            <artifactId>jjwt-api</artifactId>
+            <version>0.11.5</version>
+        </dependency>
+        <dependency>
+            <groupId>io.jsonwebtoken</groupId>
+            <artifactId>jjwt-impl</artifactId>
+            <version>0.11.5</version>
+            <scope>runtime</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.jsonwebtoken</groupId>
+            <artifactId>jjwt-jackson</artifactId>
+            <version>0.11.5</version>
+            <scope>runtime</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <optional>true</optional>
+        </dependency>
+<!--        <dependency>-->
+<!--            <groupId>org.springframework.boot</groupId>-->
+<!--            <artifactId>spring-boot-starter-actuator</artifactId>-->
+<!--        </dependency>-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

websocket_service/src/main/java/com/openisle/websocket/WebsocketServiceApplication.java

@@ -0,0 +1,13 @@
+package com.openisle.websocket;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class WebsocketServiceApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(WebsocketServiceApplication.class, args);
+    }
+
+}

websocket_service/src/main/java/com/openisle/websocket/config/RabbitMQConfig.java

@@ -0,0 +1,27 @@
+package com.openisle.websocket.config;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class RabbitMQConfig {
+
+    @Bean
+    public Jackson2JsonMessageConverter messageConverter() {
+        ObjectMapper objectMapper = new ObjectMapper();
+        objectMapper.registerModule(new com.fasterxml.jackson.datatype.jsr310.JavaTimeModule());
+        objectMapper.disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+        return new Jackson2JsonMessageConverter(objectMapper);
+    }
+
+    @Bean
+    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
+        RabbitTemplate template = new RabbitTemplate(connectionFactory);
+        template.setMessageConverter(messageConverter());
+        return template;
+    }
+}

websocket_service/src/main/java/com/openisle/websocket/config/WebSocketAuthInterceptor.java

@@ -0,0 +1,84 @@
+package com.openisle.websocket.config;
+
+import com.openisle.websocket.security.JwtService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.simp.stomp.StompCommand;
+import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
+import org.springframework.messaging.support.ChannelInterceptor;
+import org.springframework.messaging.support.MessageHeaderAccessor;
+import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
+import org.springframework.security.core.context.SecurityContextHolder;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class WebSocketAuthInterceptor implements ChannelInterceptor {
+
+    private final JwtService jwtService;
+
+    @Override
+    public Message<?> preSend(Message<?> message, MessageChannel channel) {
+        StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
+
+        if (accessor != null && StompCommand.CONNECT.equals(accessor.getCommand())) {
+            log.info("WebSocket CONNECT 请求 - 开始认证");
+            
+            String authHeader = accessor.getFirstNativeHeader("Authorization");
+            log.debug("Authorization 头: {}", authHeader != null ? "存在" : "缺失");
+            
+            if (authHeader != null && authHeader.startsWith("Bearer ")) {
+                String token = authHeader.substring(7);
+                log.debug("提取的token长度: {}", token.length());
+                
+                try {
+                    String username = jwtService.extractUsername(token);
+                    log.debug("从token中提取的用户名: {}", username);
+                    
+                    if (username != null && jwtService.isTokenValid(token)) {
+                        UsernamePasswordAuthenticationToken authToken =
+                            new UsernamePasswordAuthenticationToken(username, null, new ArrayList<>());
+                        SecurityContextHolder.getContext().setAuthentication(authToken);
+                        accessor.setUser(authToken);
+                        log.info("WebSocket 连接认证成功，用户: {}", username);
+                    } else {
+                        log.warn("WebSocket 连接认证失败 - token无效或用户名为空");
+                        log.debug("用户名: {}, token有效性: {}", username, jwtService.isTokenValid(token));
+                        return null; // 拒绝连接
+                    }
+                } catch (Exception e) {
+                    log.error("WebSocket JWT token处理异常: {}", e.getMessage(), e);
+                    return null; // 拒绝连接
+                }
+            } else {
+                log.warn("WebSocket 连接认证失败 - 缺少有效的Authorization头");
+                log.debug("Authorization头内容: {}", authHeader);
+                return null; // 拒绝连接
+            }
+        } else if (accessor != null && StompCommand.SUBSCRIBE.equals(accessor.getCommand())) {
+            log.debug("WebSocket SUBSCRIBE 请求到: {}", accessor.getDestination());
+        } else if (accessor != null && StompCommand.SEND.equals(accessor.getCommand())) {
+            log.debug("WebSocket SEND 请求到: {}", accessor.getDestination());
+        }
+        
+        return message;
+    }
+
+    @Override
+    public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
+        StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
+        
+        if (accessor != null) {
+            if (StompCommand.CONNECT.equals(accessor.getCommand()) && sent) {
+                log.info("WebSocket 连接建立成功");
+            } else if (StompCommand.DISCONNECT.equals(accessor.getCommand())) {
+                log.info("WebSocket 连接已断开");
+            }
+        }
+    }
+}

websocket_service/src/main/java/com/openisle/websocket/config/WebSocketConfig.java

@@ -0,0 +1,73 @@
+package com.openisle.websocket.config;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.simp.config.ChannelRegistration;
+import org.springframework.messaging.simp.config.MessageBrokerRegistry;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
+import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
+import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
+
+@Configuration
+@EnableWebSocketMessageBroker
+@RequiredArgsConstructor
+public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
+
+    private final WebSocketAuthInterceptor webSocketAuthInterceptor;
+    
+    @Value("${app.website-url}")
+    private String websiteUrl;
+
+    @Override
+    public void configureMessageBroker(MessageBrokerRegistry config) {
+        ThreadPoolTaskScheduler ts = new ThreadPoolTaskScheduler();
+        ts.setPoolSize(1);
+        ts.setThreadNamePrefix("wss-heartbeat-thread-");
+        ts.initialize();
+
+        config.enableSimpleBroker("/queue", "/topic")
+              .setHeartbeatValue(new long[]{10000, 10000})
+              .setTaskScheduler(ts);
+        config.setApplicationDestinationPrefixes("/app");
+        config.setUserDestinationPrefix("/user");
+    }
+
+    @Override
+    public void registerStompEndpoints(StompEndpointRegistry registry) {
+        // 1) 原生 WebSocket（不带 SockJS）
+        registry.addEndpoint("/api/ws")
+                .setAllowedOriginPatterns(
+                        "https://staging.open-isle.com",
+                        "https://www.staging.open-isle.com",
+                        websiteUrl,
+                        websiteUrl.replace("://www.", "://"),
+                        "http://localhost:*",
+                        "http://127.0.0.1:*",
+                        "http://192.168.7.98:*",
+                        "http://30.211.97.238:*"
+                );
+
+        // 2) SockJS 回退：单独路径
+        registry.addEndpoint("/api/sockjs")
+                .setAllowedOriginPatterns(
+                        "https://staging.open-isle.com",
+                        "https://www.staging.open-isle.com",
+                        websiteUrl,
+                        websiteUrl.replace("://www.", "://"),
+                        "http://localhost:*",
+                        "http://127.0.0.1:*",
+                        "http://192.168.7.98:*",
+                        "http://30.211.97.238:*"
+                )
+                .withSockJS()
+                .setWebSocketEnabled(true)
+                .setSessionCookieNeeded(false);
+    }
+
+    @Override
+    public void configureClientInboundChannel(ChannelRegistration registration) {
+        registration.interceptors(webSocketAuthInterceptor);
+    }
+}

websocket_service/src/main/java/com/openisle/websocket/dto/MessageNotificationPayload.java

@@ -0,0 +1,15 @@
+package com.openisle.websocket.dto;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.io.Serializable;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class MessageNotificationPayload implements Serializable {
+    private String targetUsername;
+    private Object payload;
+}

websocket_service/src/main/java/com/openisle/websocket/listener/NotificationListener.java

@@ -0,0 +1,114 @@
+package com.openisle.websocket.listener;
+
+import com.openisle.websocket.dto.MessageNotificationPayload;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.simp.SimpMessagingTemplate;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.Map;
+
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class NotificationListener {
+
+    private final SimpMessagingTemplate messagingTemplate;
+
+    /**
+     * Unified listener for all sharded queues and the backward-compatible legacy queue.
+     *
+     * @param payload The message payload.
+     * @param queueName The name of the queue the message was consumed from. This header is optional.
+     */
+    @RabbitListener(
+            id = "shardedListenerContainer",
+            queues = {
+                "notifications-queue-0", "notifications-queue-1", "notifications-queue-2", "notifications-queue-3",
+                "notifications-queue-4", "notifications-queue-5", "notifications-queue-6", "notifications-queue-7",
+                "notifications-queue-8", "notifications-queue-9", "notifications-queue-a", "notifications-queue-b",
+                "notifications-queue-c", "notifications-queue-d", "notifications-queue-e", "notifications-queue-f",
+                "notifications-queue"
+            }
+    )
+    public void receiveMessage(MessageNotificationPayload payload, @Header("amqp_consumedQueue") @Nullable String queueName) {
+        if (queueName != null) {
+            String queueNamePrefix = "notifications-queue-";
+            if (queueName.startsWith(queueNamePrefix)) {
+                String shardIndexStr = queueName.substring(queueNamePrefix.length());
+                log.info("=== RabbitMQ Message Received from Shard {} ({}) ===", shardIndexStr, queueName);
+            } else {
+                log.info("=== RabbitMQ Message Received from Legacy Queue ({}) ===", queueName);
+            }
+        }
+        String username = payload.getTargetUsername();
+        Object payloadObject = payload.getPayload();
+        log.info("Target username: {}", username);
+        log.info("Payload object type: {}", payloadObject != null ? payloadObject.getClass().getSimpleName() : "null");
+        log.info("Payload content: {}", payloadObject);
+
+        try {
+            if (payloadObject instanceof Map) {
+                Map<String, Object> payloadMap = (Map<String, Object>) payloadObject;
+
+                // 处理包含完整对话信息的消息 - 完全复制之前的WebSocket发送逻辑
+                if (payloadMap.containsKey("message") && payloadMap.containsKey("conversation") && payloadMap.containsKey("senderId")) {
+                    Object messageObj = payloadMap.get("message");
+                    Map<String, Object> conversationInfo = (Map<String, Object>) payloadMap.get("conversation");
+                    Long conversationId = ((Number) conversationInfo.get("id")).longValue();
+                    Long senderId = ((Number) payloadMap.get("senderId")).longValue();
+                    List<Map<String, Object>> participants = (List<Map<String, Object>>) conversationInfo.get("participants");
+
+                    // 1. 发送到conversation topic
+                    String conversationDestination = "/topic/conversation/" + conversationId;
+                    messagingTemplate.convertAndSend(conversationDestination, messageObj);
+                    log.info("Message broadcasted to destination: {}", conversationDestination);
+
+                    // 2. 为所有参与者（除发送者外）发送到个人频道和未读数量
+                    for (Map<String, Object> participant : participants) {
+                        Long participantUserId = ((Number) participant.get("userId")).longValue();
+                        String participantUsername = (String) participant.get("username");
+
+                        if (!participantUserId.equals(senderId)) {
+                            // 发送到用户个人消息频道
+                            String userDestination = "/topic/user/" + participantUserId + "/messages";
+                            messagingTemplate.convertAndSend(userDestination, messageObj);
+                            log.info("Message notification sent to destination: {}", userDestination);
+
+                            // 优先从 participant 中获取未读信息，兼容旧格式
+                            Object unreadCount = participant.getOrDefault("unreadCount", payloadMap.get("unreadCount"));
+                            if (unreadCount != null) {
+                                messagingTemplate.convertAndSendToUser(participantUsername, "/queue/unread-count", unreadCount);
+                                log.info("Sent unread count to user {} via /user/{}/queue/unread-count", participantUsername, participantUsername);
+                            }
+
+                            Object channelUnread = participant.getOrDefault("channelUnread", payloadMap.get("channelUnread"));
+                            if (channelUnread != null) {
+                                messagingTemplate.convertAndSendToUser(participantUsername, "/queue/channel-unread", channelUnread);
+                                log.info("Sent channel-unread to {}", participantUsername);
+                            }
+                        }
+                    }
+                }
+                // 处理简化的消息格式（向后兼容）
+                else if (payloadMap.containsKey("message")) {
+
+                    if (payloadMap.containsKey("unreadCount")) {
+                        messagingTemplate.convertAndSendToUser(username, "/queue/unread-count", payloadMap.get("unreadCount"));
+                        log.info("Sent unread count to user {} via /user/{}/queue/unread-count", username, username);
+                    }
+                    if (payloadMap.containsKey("channelUnread")) {
+                        messagingTemplate.convertAndSendToUser(username, "/queue/channel-unread", payloadMap.get("channelUnread"));
+                        log.info("Sent channel-unread to {}", username);
+                    }
+                }
+            }
+        } catch (Exception e) {
+            log.error("Failed to process and send message for user {}", username, e);
+        }
+    }
+}

websocket_service/src/main/java/com/openisle/websocket/security/ApplicationConfig.java

@@ -0,0 +1,19 @@
+package com.openisle.websocket.security;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.security.core.userdetails.User;
+import org.springframework.security.core.userdetails.UserDetailsService;
+
+import java.util.Collections;
+
+@Configuration
+@RequiredArgsConstructor
+public class ApplicationConfig {
+
+    @Bean
+    public UserDetailsService userDetailsService() {
+        return username -> new User(username, "", Collections.emptyList());
+    }
+}

websocket_service/src/main/java/com/openisle/websocket/security/JwtService.java

@@ -0,0 +1,98 @@
+package com.openisle.websocket.security;
+
+import io.jsonwebtoken.Claims;
+import io.jsonwebtoken.Jwts;
+import io.jsonwebtoken.security.Keys;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.StandardCharsets;
+import java.security.Key;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Date;
+import java.util.function.Function;
+
+@Service
+public class JwtService {
+    
+    private static final Logger logger = LoggerFactory.getLogger(JwtService.class);
+
+    @Value("${app.jwt.secret}")
+    private String secret;
+
+    public String extractUsername(String token) {
+        return extractClaim(token, Claims::getSubject);
+    }
+
+    public <T> T extractClaim(String token, Function<Claims, T> claimsResolver) {
+        final Claims claims = extractAllClaims(token);
+        return claimsResolver.apply(claims);
+    }
+
+    public boolean isTokenValid(String token) {
+        try {
+            return !isTokenExpired(token);
+        } catch (Exception e) {
+            return false;
+        }
+    }
+
+    private boolean isTokenExpired(String token) {
+        return extractExpiration(token).before(new Date());
+    }
+
+    private Date extractExpiration(String token) {
+        return extractClaim(token, Claims::getExpiration);
+    }
+
+    private Claims extractAllClaims(String token) {
+        logger.debug("解析JWT token - secret长度: {}", secret != null ? secret.length() : "null");
+        
+        try {
+            return Jwts
+                    .parserBuilder()
+                    .setSigningKey(getSignInKey())
+                    .build()
+                    .parseClaimsJws(token)
+                    .getBody();
+        } catch (Exception e) {
+            logger.error("JWT解析失败: {}", e.getMessage());
+            throw e;
+        }
+    }
+
+    private Key getSignInKey() {
+        // 使用与backend相同的密钥处理方式：直接Base64解码
+        byte[] keyBytes;
+        try {
+            // 尝试Base64解码
+            keyBytes = java.util.Base64.getDecoder().decode(secret);
+        } catch (IllegalArgumentException e) {
+            // 如果不是Base64格式，使用UTF-8字节
+            keyBytes = secret.getBytes(StandardCharsets.UTF_8);
+            // 确保密钥长度至少256位（32字节）
+            if (keyBytes.length < 32) {
+                MessageDigest digest;
+                try {
+                    digest = MessageDigest.getInstance("SHA-256");
+                    keyBytes = digest.digest(keyBytes);
+                } catch (NoSuchAlgorithmException ex) {
+                    throw new IllegalStateException("SHA-256 not available", ex);
+                }
+            }
+        }
+        return Keys.hmacShaKeyFor(keyBytes);
+    }
+
+    public String validateAndGetSubject(String token) {
+        Claims claims = Jwts.parserBuilder()
+                .setSigningKey(getSignInKey())
+                .build()
+                .parseClaimsJws(token)
+                .getBody();
+        return claims.getSubject();
+    }
+}

websocket_service/src/main/java/com/openisle/websocket/security/SecurityConfig.java

@@ -0,0 +1,71 @@
+package com.openisle.websocket.security;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.security.config.Customizer;
+import org.springframework.security.config.annotation.web.builders.HttpSecurity;
+import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity;
+import org.springframework.security.config.http.SessionCreationPolicy;
+import org.springframework.security.web.SecurityFilterChain;
+import org.springframework.web.cors.CorsConfiguration;
+import org.springframework.web.cors.CorsConfigurationSource;
+import org.springframework.web.cors.UrlBasedCorsConfigurationSource;
+
+import java.util.List;
+
+@Configuration
+@EnableWebSecurity
+@RequiredArgsConstructor
+public class SecurityConfig {
+
+    @Value("${app.website-url}")
+    private String websiteUrl;
+
+    @Bean
+    public CorsConfigurationSource corsConfigurationSource() {
+        CorsConfiguration configuration = new CorsConfiguration();
+        configuration.setAllowedOrigins(List.of(
+                "http://127.0.0.1:8080",
+                "http://127.0.0.1:8081",
+                "http://127.0.0.1:8082",
+                "http://127.0.0.1:3000",
+                "http://127.0.0.1:3001",
+                "http://127.0.0.1",
+                "http://localhost:8080",
+                "http://localhost:8081",
+                "http://localhost:8082",
+                "http://localhost:3000",
+                "http://localhost:3001",
+                "http://localhost",
+                "http://30.211.97.238:3000",
+                "http://30.211.97.238",
+                "http://192.168.7.98",
+                "http://192.168.7.98:3000",
+                websiteUrl,
+                websiteUrl.replace("://www.", "://")
+        ));
+        configuration.setAllowedMethods(List.of("GET", "POST", "PUT", "DELETE", "OPTIONS"));
+        configuration.setAllowedHeaders(List.of("*"));
+        configuration.setAllowCredentials(true);
+
+        UrlBasedCorsConfigurationSource source = new UrlBasedCorsConfigurationSource();
+        source.registerCorsConfiguration("/**", configuration);
+        return source;
+    }
+
+    @Bean
+    public SecurityFilterChain securityFilterChain(HttpSecurity http) throws Exception {
+        http
+                .csrf().disable()
+                .cors(Customizer.withDefaults())
+                .authorizeHttpRequests(auth -> auth
+                        .requestMatchers("/**").permitAll() // Permit all HTTP requests
+                )
+                .sessionManagement(session -> session
+                        .sessionCreationPolicy(SessionCreationPolicy.STATELESS)
+                );
+        return http.build();
+    }
+}

websocket_service/src/main/resources/application.properties

@@ -0,0 +1,22 @@
+server.port=${SERVER_PORT:8082}
+
+# 服务器配置
+spring.application.name=websocket-service
+
+# RabbitMQ 配置
+spring.rabbitmq.host=${RABBITMQ_HOST:localhost}
+spring.rabbitmq.port=${RABBITMQ_PORT:5672}
+spring.rabbitmq.username=${RABBITMQ_USERNAME:guest}
+spring.rabbitmq.password=${RABBITMQ_PASSWORD:guest}
+spring.rabbitmq.virtual-host=/
+
+# JWT 配置
+app.jwt.secret=${JWT_SECRET:jwt_sec}
+
+# 日志配置
+logging.level.com.openisle=${LOG_LEVEL:INFO}
+logging.level.org.springframework.messaging=${MESSAGING_LOG_LEVEL:DEBUG}
+logging.level.org.springframework.web.socket=${WEBSOCKET_LOG_LEVEL:DEBUG}
+
+# 网站 URL 配置
+app.website-url=${WEBSITE_URL:https://www.open-isle.com}

websocket_service/src/main/resources/logback-spring.xml

@@ -0,0 +1,140 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <!-- 定义日志输出格式 -->
+    <property name="LOG_PATTERN" value="%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n"/>
+    
+    <!-- 控制台输出 -->
+    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>${LOG_PATTERN}</pattern>
+            <charset>UTF-8</charset>
+        </encoder>
+    </appender>
+    
+    <!-- 文件输出 -->
+    <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <file>logs/websocket-service.log</file>
+        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
+            <fileNamePattern>logs/websocket-service.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
+            <maxFileSize>10MB</maxFileSize>
+            <maxHistory>30</maxHistory>
+            <totalSizeCap>1GB</totalSizeCap>
+        </rollingPolicy>
+        <encoder>
+            <pattern>${LOG_PATTERN}</pattern>
+            <charset>UTF-8</charset>
+        </encoder>
+    </appender>
+    
+    <!-- WebSocket 相关日志 -->
+    <appender name="WEBSOCKET_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <file>logs/websocket.log</file>
+        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
+            <fileNamePattern>logs/websocket.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
+            <maxFileSize>10MB</maxFileSize>
+            <maxHistory>7</maxHistory>
+            <totalSizeCap>500MB</totalSizeCap>
+        </rollingPolicy>
+        <encoder>
+            <pattern>${LOG_PATTERN}</pattern>
+            <charset>UTF-8</charset>
+        </encoder>
+    </appender>
+    
+    <!-- 错误日志单独输出 -->
+    <appender name="ERROR_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <filter class="ch.qos.logback.classic.filter.LevelFilter">
+            <level>ERROR</level>
+            <onMatch>ACCEPT</onMatch>
+            <onMismatch>DENY</onMismatch>
+        </filter>
+        <file>logs/error.log</file>
+        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
+            <fileNamePattern>logs/error.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
+            <maxFileSize>10MB</maxFileSize>
+            <maxHistory>30</maxHistory>
+            <totalSizeCap>500MB</totalSizeCap>
+        </rollingPolicy>
+        <encoder>
+            <pattern>${LOG_PATTERN}</pattern>
+            <charset>UTF-8</charset>
+        </encoder>
+    </appender>
+    
+    <!-- 异步日志配置 -->
+    <appender name="ASYNC_FILE" class="ch.qos.logback.classic.AsyncAppender">
+        <discardingThreshold>0</discardingThreshold>
+        <queueSize>512</queueSize>
+        <includeCallerData>false</includeCallerData>
+        <appender-ref ref="FILE"/>
+    </appender>
+    
+    <!-- 异步 WebSocket 日志 -->
+    <appender name="ASYNC_WEBSOCKET" class="ch.qos.logback.classic.AsyncAppender">
+        <discardingThreshold>0</discardingThreshold>
+        <queueSize>256</queueSize>
+        <includeCallerData>false</includeCallerData>
+        <appender-ref ref="WEBSOCKET_FILE"/>
+    </appender>
+    
+    <!-- 特定包的日志级别配置 -->
+    <logger name="com.openisle.websocket" level="INFO" additivity="false">
+        <appender-ref ref="CONSOLE"/>
+        <appender-ref ref="ASYNC_FILE"/>
+        <appender-ref ref="ERROR_FILE"/>
+    </logger>
+    
+    <!-- WebSocket 相关日志 -->
+    <logger name="com.openisle.websocket.controller.WebSocketController" level="DEBUG" additivity="false">
+        <appender-ref ref="ASYNC_WEBSOCKET"/>
+        <appender-ref ref="CONSOLE"/>
+    </logger>
+    
+    <logger name="com.openisle.websocket.config.WebSocketAuthInterceptor" level="INFO" additivity="false">
+        <appender-ref ref="ASYNC_WEBSOCKET"/>
+        <appender-ref ref="CONSOLE"/>
+    </logger>
+    
+    <logger name="com.openisle.websocket.listener.NotificationListener" level="INFO" additivity="false">
+        <appender-ref ref="ASYNC_WEBSOCKET"/>
+        <appender-ref ref="CONSOLE"/>
+    </logger>
+    
+    <!-- Spring WebSocket 日志 -->
+    <logger name="org.springframework.web.socket" level="DEBUG" additivity="false">
+        <appender-ref ref="ASYNC_WEBSOCKET"/>
+    </logger>
+    
+    <logger name="org.springframework.messaging" level="DEBUG" additivity="false">
+        <appender-ref ref="ASYNC_WEBSOCKET"/>
+    </logger>
+    
+    <!-- RabbitMQ 日志 -->
+    <logger name="org.springframework.amqp" level="INFO" additivity="false">
+        <appender-ref ref="CONSOLE"/>
+        <appender-ref ref="ASYNC_FILE"/>
+    </logger>
+    
+    <!-- 根日志级别配置 -->
+    <root level="INFO">
+        <appender-ref ref="CONSOLE"/>
+        <appender-ref ref="ASYNC_FILE"/>
+        <appender-ref ref="ERROR_FILE"/>
+    </root>
+    
+    <!-- 开发环境配置 -->
+    <springProfile name="dev">
+        <root level="DEBUG">
+            <appender-ref ref="CONSOLE"/>
+        </root>
+    </springProfile>
+    
+    <!-- 生产环境配置 -->
+    <springProfile name="prod">
+        <root level="WARN">
+            <appender-ref ref="ASYNC_FILE"/>
+            <appender-ref ref="ERROR_FILE"/>
+        </root>
+    </springProfile>
+    
+</configuration>

websocket_service/websocket_service.env.example

@@ -0,0 +1,12 @@
+SERVER_PORT=<your-server-port>
+
+# RabbitMQ 配置
+RABBITMQ_HOST=<your-host>
+RABBITMQ_PORT=<your-port>
+RABBITMQ_USERNAME=<your-username>
+RABBITMQ_PASSWORD=<your-password>
+
+# JWT 配置
+JWT_SECRET=<your-jwt-secret>
+
+WEBSITE_URL=<your-website-url>