Spring 5响应式WebSocket：单端点复用与客户端连接限制咨询

Great question—this is a common point of confusion when moving from traditional WebSocket setups to reactive ones, and you’re already on the right track by leaning into publish/subscribe patterns! Let’s break this down clearly:

Single Endpoint with Publish/Subscribe is the Way to Go #

Based on the best practices you mentioned (single connection per client, avoiding overdesign), a single endpoint handling all client requests is absolutely the better choice. Here’s why:

• Reduced overhead: Managing multiple endpoints adds unnecessary configuration and mental overhead. A single endpoint keeps your setup clean and aligned with reactive principles of efficient resource usage.
• Flexible topic-based communication: Instead of routing clients to different endpoints, you can embed a "topic" or "event type" in the WebSocket messages themselves. Using Reactor’s publish/subscribe primitives (like EmitterProcessor or TopicProcessor), you can route messages to specific client groups without needing separate connections.
• Simplified client logic: Clients only need to maintain one WebSocket connection, which is easier to manage and less resource-intensive on their end too.

How to Implement This with Your Existing Setup #

Your current AppConfig is already set up for a single endpoint (/event-emitter)—you just need to extend your DomWebSocketHandler to handle topic-based subscriptions:

1. Define global publish/subscribe processors: Create shared processors for each event topic your app needs (e.g., orderUpdatesProcessor, notificationProcessor). These act as message brokers between your backend logic and connected clients.
2. Handle client subscription messages: When a client sends a message like {"topic": "order-updates", "action": "subscribe"}, your handler can subscribe the client’s WebSocketSession to the corresponding processor’s Flux.
3. Clean up on session close: Make sure to unsubscribe clients from processors when their WebSocket session closes to avoid memory leaks.

Here’s a simplified snippet for your handler:

import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketSession;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.EmitterProcessor;

public class DomWebSocketHandler implements WebSocketHandler {
    // Shared processors for different topics
    private final EmitterProcessor<String> orderUpdatesProcessor = EmitterProcessor.create();
    private final EmitterProcessor<String> notificationProcessor = EmitterProcessor.create();

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        // Parse incoming messages to handle subscriptions
        Flux<String> incoming = session.receive()
                .map(msg -> msg.getPayloadAsText())
                .doOnNext(msg -> {
                    // Example: Parse message to get topic and action
                    if (msg.contains("\"topic\": \"order-updates\"")) {
                        orderUpdatesProcessor.subscribe(update -> 
                            session.sendText(Mono.just(update)).subscribe()
                        );
                    } else if (msg.contains("\"topic\": \"notifications\"")) {
                        notificationProcessor.subscribe(notification -> 
                            session.sendText(Mono.just(notification)).subscribe()
                        );
                    }
                });

        // Combine incoming and outgoing streams, clean up on close
        return incoming.then(session.close())
                .doFinally(signalType -> {
                    // Optional: Add explicit unsubscription logic if needed
                });
    }

    // Method to send messages to a topic from your backend
    public void sendOrderUpdate(String update) {
        orderUpdatesProcessor.onNext(update);
    }
}

How to Limit Single Client Connections #

To enforce one connection per client (e.g., per user ID or session ID), you can maintain a map of active connections tied to a unique client identifier:

1. Track active connections: Use a thread-safe map like ConcurrentHashMap<String, WebSocketSession> to store client IDs and their active sessions.
2. Validate new connections: When a new session is established, extract the client ID (from headers, JWT, or initial message) and check if an active connection already exists. If it does, close the new session (or optionally terminate the old one).
3. Clean up on session close: Remove the client ID from the map when the session closes to keep the map accurate.

Example addition to your handler:

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class DomWebSocketHandler implements WebSocketHandler {
    private final Map<String, WebSocketSession> activeConnections = new ConcurrentHashMap<>();

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        // Extract client ID (example: from a custom header)
        String clientId = session.getHandshakeInfo().getHeaders().getFirst("X-Client-ID");
        
        if (clientId != null && activeConnections.containsKey(clientId)) {
            // Close new connection if client already has an active one
            return session.close();
        }
        if (clientId != null) {
            activeConnections.put(clientId, session);
        }

        // ... existing logic for subscriptions ...

        return incoming.then(session.close())
                .doFinally(signalType -> {
                    if (clientId != null) {
                        activeConnections.remove(clientId);
                    }
                });
    }
}

When Would Multiple Endpoints Make Sense? #

While single endpoint is preferred for most cases, multiple endpoints might be justified if:

• Your use cases are completely isolated (e.g., one endpoint for admin events, another for user-facing notifications) and you want strict separation of concerns.
• Your team is more comfortable with REST-like endpoint partitioning for organizational purposes.

Even then, you can still use publish/subscribe under the hood—multiple endpoints would just act as entry points for different client groups. But for most reactive WebSocket setups, the single endpoint approach is more efficient and aligns better with best practices.

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