<!-- Spring WebSocket -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<!-- Redis (用于會(huì)話存儲(chǔ)和消息轉(zhuǎn)發(fā)) -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- 連接池 (可選，優(yōu)化Redis性能) -->
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
</dependency>

spring:
  redis:
    host: localhost  # Redis服務(wù)器地址
    port: 6379       # Redis端口
    password:        # 密碼（如有）
    lettuce:
      pool:
        max-active: 8  # 最大連接數(shù)
        max-idle: 8    # 最大空閑連接
        min-idle: 2    # 最小空閑連接

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
import org.springframework.web.socket.server.HandshakeInterceptor;
@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
    private final WebSocketHandler webSocketHandler;
    private final HandshakeInterceptor handshakeInterceptor;
    // 注入自定義處理器和攔截器
    public WebSocketConfig(WebSocketHandler webSocketHandler, HandshakeInterceptor handshakeInterceptor) {
        this.webSocketHandler = webSocketHandler;
        this.handshakeInterceptor = handshakeInterceptor;
    }
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        // 注冊(cè)WebSocket端點(diǎn)，允許跨域
        registry.addHandler(webSocketHandler, "/ws")
                .addInterceptors(handshakeInterceptor)
                .setAllowedOrigins("*"); // 生產(chǎn)環(huán)境需指定具體域名
    }
}

public class RedisKeyConstants {
    // 存儲(chǔ)用戶-節(jié)點(diǎn)映射：key=userId，value=nodeId
    public static final String USER_NODE_MAP = "websocket:user:node";
    // 存儲(chǔ)節(jié)點(diǎn)信息：key=nodeId，value=節(jié)點(diǎn)信息（如IP:端口）
    public static final String NODE_INFO = "websocket:node:info";
    // Redis Pub/Sub頻道（用于節(jié)點(diǎn)間消息轉(zhuǎn)發(fā)）
    public static final String MSG_CHANNEL = "websocket:msg:channel";
}

import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.http.server.ServletServerHttpRequest;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.HandshakeInterceptor;
import org.springframework.data.redis.core.StringRedisTemplate;
import javax.servlet.http.HttpSession;
import java.util.Map;
import java.util.UUID;
@Component
public class WebSocketHandshakeInterceptor implements HandshakeInterceptor {
    private final StringRedisTemplate redisTemplate;
    private final String nodeId; // 當(dāng)前節(jié)點(diǎn)唯一標(biāo)識(shí)（如UUID）
    public WebSocketHandshakeInterceptor(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.nodeId = UUID.randomUUID().toString(); // 節(jié)點(diǎn)啟動(dòng)時(shí)生成唯一ID
    }
    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response,
                                   WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {
        // 從請(qǐng)求參數(shù)或Session中獲取用戶ID（根據(jù)業(yè)務(wù)場(chǎng)景調(diào)整）
        ServletServerHttpRequest servletRequest = (ServletServerHttpRequest) request;
        HttpSession session = servletRequest.getServletRequest().getSession();
        String userId = (String) session.getAttribute("userId"); // 假設(shè)用戶已登錄，Session中存userId
        if (userId != null) {
            // 將用戶與當(dāng)前節(jié)點(diǎn)關(guān)聯(lián)（存儲(chǔ)到Redis）
            redisTemplate.opsForValue().set(
                    RedisKeyConstants.USER_NODE_MAP + ":" + userId,
                    nodeId
            );
            // 注冊(cè)當(dāng)前節(jié)點(diǎn)信息（可選，用于監(jiān)控）
            redisTemplate.opsForValue().set(
                    RedisKeyConstants.NODE_INFO + ":" + nodeId,
                    request.getLocalAddress().toString() // 節(jié)點(diǎn)地址
            );
            attributes.put("userId", userId); // 傳遞userId到處理器
            return true;
        }
        return false; // 未登錄用戶拒絕連接
    }
    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response,
                               WebSocketHandler wsHandler, Exception exception) {}
}

import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Component
public class WebSocketHandler extends TextWebSocketHandler {
    // 本地會(huì)話緩存（當(dāng)前節(jié)點(diǎn)的WebSocket會(huì)話）
    private final Map<String, WebSocketSession> localSessions = new ConcurrentHashMap<>();
    private final StringRedisTemplate redisTemplate;
    private final ObjectMapper objectMapper;
    private final String nodeId; // 當(dāng)前節(jié)點(diǎn)ID（與攔截器一致）
    public WebSocketHandler(StringRedisTemplate redisTemplate, ObjectMapper objectMapper,
                           @Value("${websocket.node.id}") String nodeId) {
        this.redisTemplate = redisTemplate;
        this.objectMapper = objectMapper;
        this.nodeId = nodeId;
        // 訂閱Redis頻道，接收其他節(jié)點(diǎn)的消息
        subscribeToRedisChannel();
    }
    // 連接建立時(shí)，緩存本地會(huì)話
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        String userId = (String) session.getAttributes().get("userId");
        localSessions.put(userId, session);
    }
    // 接收客戶端消息
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        String userId = (String) session.getAttributes().get("userId");
        String payload = message.getPayload(); // 客戶端發(fā)送的消息
        // 解析消息（假設(shè)消息格式：{"targetUserId": "xxx", "content": "xxx"}）
        Map<String, String> msg = objectMapper.readValue(payload, Map.class);
        String targetUserId = msg.get("targetUserId");
        String content = msg.get("content");
        // 發(fā)送消息給目標(biāo)用戶（跨節(jié)點(diǎn)則轉(zhuǎn)發(fā)）
        sendMessageToUser(targetUserId, content);
    }
    // 發(fā)送消息給指定用戶
    public void sendMessageToUser(String targetUserId, String content) throws IOException {
        // 1. 從Redis獲取目標(biāo)用戶所在節(jié)點(diǎn)
        String targetNodeId = redisTemplate.opsForValue().get(
                RedisKeyConstants.USER_NODE_MAP + ":" + targetUserId
        );
        if (targetNodeId == null) {
            throw new RuntimeException("用戶未連接");
        }
        // 2. 若目標(biāo)節(jié)點(diǎn)是當(dāng)前節(jié)點(diǎn)，直接發(fā)送；否則通過(guò)Redis轉(zhuǎn)發(fā)
        if (targetNodeId.equals(nodeId)) {
            WebSocketSession session = localSessions.get(targetUserId);
            if (session != null && session.isOpen()) {
                session.sendMessage(new TextMessage(content));
            }
        } else {
            // 構(gòu)造跨節(jié)點(diǎn)消息（包含目標(biāo)用戶、內(nèi)容、目標(biāo)節(jié)點(diǎn)）
            Map<String, String> crossMsg = new HashMap<>();
            crossMsg.put("targetUserId", targetUserId);
            crossMsg.put("content", content);
            crossMsg.put("targetNodeId", targetNodeId);
            // 發(fā)布到Redis頻道
            redisTemplate.convertAndSend(
                    RedisKeyConstants.MSG_CHANNEL,
                    objectMapper.writeValueAsString(crossMsg)
            );
        }
    }
    // 訂閱Redis頻道，接收其他節(jié)點(diǎn)的消息并轉(zhuǎn)發(fā)給本地用戶
    private void subscribeToRedisChannel() {
        redisTemplate.getConnectionFactory().getConnection().subscribe(
                message -> {
                    try {
                        String payload = new String(message.getBody());
                        Map<String, String> crossMsg = objectMapper.readValue(payload, Map.class);
                        // 僅處理目標(biāo)節(jié)點(diǎn)為當(dāng)前節(jié)點(diǎn)的消息
                        if (crossMsg.get("targetNodeId").equals(nodeId)) {
                            String targetUserId = crossMsg.get("targetUserId");
                            String content = crossMsg.get("content");
                            WebSocketSession session = localSessions.get(targetUserId);
                            if (session != null && session.isOpen()) {
                                session.sendMessage(new TextMessage(content));
                            }
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                },
                RedisKeyConstants.MSG_CHANNEL.getBytes()
        );
    }
    // 連接關(guān)閉時(shí)，清理會(huì)話和Redis映射
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        String userId = (String) session.getAttributes().get("userId");
        localSessions.remove(userId);
        // 刪除Redis中的用戶-節(jié)點(diǎn)映射
        redisTemplate.delete(RedisKeyConstants.USER_NODE_MAP + ":" + userId);
    }
}

websocket:
  node:
    id: ${HOSTNAME:node-${random.uuid}} # 優(yōu)先使用主機(jī)名，否則隨機(jī)生成