如何在Spring WebFlux中通过WebSocket向特定客户端发送消息？

嘿，我来帮你搞定这个问题！结合你用React/Redux开发客户端、Spring WebFlux（搭配Reactor和Spring Boot 2）做服务端的场景，咱们一步步实现给特定客户端发消息的功能：

核心思路 #

要给指定客户端发消息，关键是跟踪每个客户端的WebSocket连接——咱们需要用一个唯一标识（比如用户ID、客户端会话ID）来关联对应的WebSocketSession，这样就能精准定位到要发送的目标客户端。

具体实现步骤 #

1. 实现WebSocket处理器，维护客户端连接映射 #

创建一个WebSocketHandler的实现类，这里咱们用线程安全的Map来存储“客户端标识-WebSocket会话”的对应关系，同时处理连接的建立、关闭和消息接收：

import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketSession;
import reactor.core.publisher.Mono;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class ChatWebSocketHandler implements WebSocketHandler {

    // 用ConcurrentHashMap保证多线程环境下的安全性，key是客户端唯一标识（比如用户ID）
    private final Map<String, WebSocketSession> activeSessions = new ConcurrentHashMap<>();

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        // 从握手URL的参数中获取客户端标识（客户端连接时需要传这个参数）
        String clientId = extractClientIdFromHandshake(session);
        
        // 连接建立时，将会话存入映射
        activeSessions.put(clientId, session);
        System.out.printf("客户端 [%s] 已连接%n", clientId);

        // 处理客户端发来的消息（根据你的业务需求调整，可选）
        return session.receive()
                .doOnNext(message -> {
                    String payload = message.getPayloadAsText();
                    System.out.printf("收到客户端 [%s] 的消息: %s%n", clientId, payload);
                })
                .doFinally(signalType -> {
                    // 连接关闭时，从映射中移除会话，避免内存泄漏
                    activeSessions.remove(clientId);
                    System.out.printf("客户端 [%s] 已断开连接%n", clientId);
                })
                .then();
    }

    // 提取客户端标识的工具方法
    private String extractClientIdFromHandshake(WebSocketSession session) {
        // 假设客户端连接时通过URL参数传递clientId，比如ws://xxx/websocket/chat?clientId=123
        String query = session.getHandshakeInfo().getUri().getQuery();
        if (query != null && query.contains("clientId=")) {
            return query.split("=")[1];
        }
        // 如果没有传，也可以用会话ID作为临时标识
        return session.getId();
    }

    // 对外暴露的方法：给指定客户端发送消息
    public Mono<Void> sendMessageToClient(String clientId, String messageContent) {
        WebSocketSession targetSession = activeSessions.get(clientId);
        if (targetSession != null && targetSession.isOpen()) {
            // 发送文本消息
            return targetSession.send(Mono.just(targetSession.textMessage(messageContent)));
        }
        // 如果客户端不存在或已断开，返回空Mono表示无操作
        return Mono.empty();
    }
}

2. 配置WebSocket路由 #

把咱们的处理器注册到WebFlux的路由中，让服务端能接收WebSocket连接请求：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.HandlerMapping;
import org.springframework.web.reactive.handler.SimpleUrlHandlerMapping;
import org.springframework.web.reactive.socket.server.support.WebSocketHandlerAdapter;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class WebSocketConfig {

    @Bean
    public HandlerMapping webSocketRoute(ChatWebSocketHandler chatWebSocketHandler) {
        Map<String, WebSocketHandler> routeMap = new HashMap<>();
        // 这里的路径要和你客户端连接的路径完全一致：/websocket/chat
        routeMap.put("/websocket/chat", chatWebSocketHandler);

        SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
        mapping.setUrlMap(routeMap);
        mapping.setOrder(-1); // 确保这个路由优先被处理
        return mapping;
    }

    @Bean
    public WebSocketHandlerAdapter webSocketHandlerAdapter() {
        return new WebSocketHandlerAdapter();
    }
}

3. 修改客户端代码，传递唯一标识 #

你的React客户端在连接WebSocket时，需要把客户端的唯一标识（比如当前登录用户的ID）通过URL参数传递给服务端，这样服务端才能识别目标：

// 假设从Redux store中获取当前用户ID
const currentUserId = this.props.currentUser.id;
// 拼接带clientId参数的WebSocket连接地址
this.clientWebSocket = new WebSocket(`ws://127.0.0.1:8080/websocket/chat?clientId=${currentUserId}`);

// 保持你原来的连接回调逻辑
this.clientWebSocket.onopen = function() {
    console.log("WebSocket连接成功");
}
this.clientWebSocket.onclose = function() {
    console.log("WebSocket连接关闭");
}
this.clientWebSocket.onerror = function(error) {
    console.error("WebSocket连接出错:", error);
}
const _this = this;
this.clientWebSocket.onmessage = function(dataFromServer) {
    // 处理服务端发来的消息，更新你的消息列表
    _this.updateMessages(dataFromServer.data);
}

4. 在业务逻辑中调用发消息方法 #

比如你有一个业务服务需要给特定用户发送通知，只需注入咱们的ChatWebSocketHandler，调用sendMessageToClient方法即可：

import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

@Service
public class ChatNotificationService {

    private final ChatWebSocketHandler chatWebSocketHandler;

    // 构造方法注入处理器
    public ChatNotificationService(ChatWebSocketHandler chatWebSocketHandler) {
        this.chatWebSocketHandler = chatWebSocketHandler;
    }

    // 示例：给指定用户发送系统通知
    public Mono<Void> sendSystemNotification(String userId, String notificationContent) {
        String formattedMessage = String.format("[系统通知] %s", notificationContent);
        return chatWebSocketHandler.sendMessageToClient(userId, formattedMessage);
    }
}

关键注意事项 #

• 线程安全：必须用ConcurrentHashMap存储会话，因为WebFlux是异步非阻塞模型，多个线程可能同时操作会话映射。
• 连接清理：一定要在连接关闭时移除会话，避免内存泄漏。
• 标识安全性：生产环境中，建议从JWT令牌或认证头中解析客户端标识，而不是直接用URL参数，防止伪造。

内容的提问来源于stack exchange，提问作者RuF