非STOMP客户端适配Spring WebSocket及相关技术疑问

问题1：当前实现方案的合理性及替代方案 #

你的判断是对的，当前在服务器端给自己建立STOMP连接的方案存在明显缺陷：

• 额外资源开销：每个客户端连接都会在服务器内部创建一个STOMP客户端连接，相当于双倍的连接资源占用，并发高时会显著增加内存和线程消耗。
• 不必要的消息转发：消息从Broker到内部STOMP客户端，再转到外部WebSocket会话，多了一次转发链路，增加延迟和故障点。

更好的替代方案是直接基于Spring WebSocket原生能力，手动实现订阅/发布逻辑，无需依赖STOMP协议层：

1. 自定义订阅协议：定义简单的文本格式（比如JSON）让客户端发送订阅/取消订阅指令，例如{"action":"subscribe","topic":"topic1"}。
2. 维护订阅关系映射：用线程安全的Map<String, Set<WebSocketSession>>（key为主题，value为订阅该主题的会话集合），在handleTextMessage中解析客户端指令，添加或移除会话。
3. 直接发送消息：在事件处理器中，通过遍历对应主题的会话集合调用session.sendMessage发送消息，或者结合SimpMessagingTemplate实现路由。

简化示例代码：

@Component
class MyWebSocketHandler : TextWebSocketHandler() {
    // 线程安全的主题-会话映射
    private val topicSubscriptions = ConcurrentHashMap<String, ConcurrentHashMap<String, WebSocketSession>>()

    override fun afterConnectionEstablished(session: WebSocketSession) {
        // 会话初始化逻辑
    }

    override fun handleTextMessage(session: WebSocketSession, message: TextMessage) {
        val payload = message.payload
        if (isSubscribeMessage(payload)) {
            val topicId = extractTopicIdSubscribeMessage(payload)
            topicSubscriptions.computeIfAbsent(topicId) { ConcurrentHashMap() }
                .put(session.id, ConcurrentWebSocketSessionDecorator(session, 20_000, 10_000_000))
        } else if (isUnsubscribeMessage(payload)) {
            val topicId = extractTopicIdUnsubscribeMessage(payload)
            topicSubscriptions[topicId]?.remove(session.id)
        }
    }

    // 供事件处理器调用的消息发送方法
    fun sendToTopic(topicId: String, message: String) {
        topicSubscriptions[topicId]?.values?.forEach { session ->
            if (session.isOpen) {
                session.sendMessage(TextMessage(message))
            }
        }
    }
}

// 改造事件处理器
class MyEventHandler(
    private val myWebSocketHandler: MyWebSocketHandler
) {
   fun handle(event: Event) {
        myWebSocketHandler.sendToTopic(event.topic, "Some notification message")
    }
}

问题2：Spring内存消息Broker vs 外部MQ（RabbitMQ/ActiveMQ）的优缺点 #

你的理解基本准确，两者核心差异体现在部署扩展性、资源隔离和可靠性：

内存Broker的优点 #

• 性能更高：无需网络IO，消息在进程内传递，延迟极低，适合小规模、低并发场景。
• 部署与配置简单：无需额外维护外部服务，Spring Boot自动完成配置。

内存Broker的缺点 #

• 无法横向扩展：多实例部署时，订阅关系和消息无法跨实例共享，只有发送消息的实例上的客户端能收到通知。
• 资源限制：订阅关系、消息队列都在JVM内存中，高并发或大消息量场景下易引发OOM或GC压力。
• 无持久化：应用重启后，所有订阅关系和未发送消息都会丢失，不适合需要消息可靠性的场景。

外部MQ的优点 #

• 支持分布式扩展：多应用实例可连接同一MQ，订阅关系和消息全局共享，适合大规模分布式部署。
• 资源隔离：消息处理、订阅管理由MQ负责，减轻应用服务器的内存和线程压力。
• 可靠性保障：支持消息持久化、重试机制，应用重启后未发送消息不会丢失。
• 高级特性：提供消息路由、死信队列、流量控制等能力，适配复杂业务场景。

选型建议 #

• 单实例部署、客户端规模较小（数千以内）、对消息可靠性要求低的场景，内存Broker足够使用。
• 需要多实例部署、高并发客户端（上万级）、或要求消息持久化的场景，必须切换到外部MQ。

问题3：Spring Webflux响应式改造与STOMP兼容方案 #

首先明确：Spring Webflux目前无官方STOMP协议支持，因STOMP是基于文本帧的命令式交互模型，而Webflux的WebSocket API是响应式流处理模型，两者编程模型差异较大。

若要保留订阅/发布的核心能力并适配Webflux，推荐以下方案：

方案1：基于Webflux原生WebSocket实现自定义订阅逻辑 #

完全抛弃STOMP协议层，用响应式方式实现自定义订阅机制，适配Webflux的流处理模型：

1. 用ConcurrentHashMap维护主题到FluxSink的映射（每个会话对应一个FluxSink，用于向客户端发送消息）。
2. 客户端连接时，建立双向消息流，解析订阅指令并将FluxSink注册到对应主题。
3. 事件处理器中，通过主题找到对应FluxSink集合，发送消息。

简化示例代码：

@Component
class ReactiveWebSocketHandler {
    private val topicSinks = ConcurrentHashMap<String, ConcurrentHashMap<String, FluxSink<String>>>()

    fun handle(session: WebSocketSession): Mono<Void> {
        val sessionId = session.id
        // 处理客户端输入消息
        val input = session.receive()
            .map { it.payloadAsText }
            .doOnNext { payload ->
                if (isSubscribeMessage(payload)) {
                    val topicId = extractTopicIdSubscribeMessage(payload)
                    topicSinks.computeIfAbsent(topicId) { ConcurrentHashMap() }
                        .computeIfAbsent(sessionId) {
                            // 创建FluxSink并绑定到会话输出流
                            val flux = Flux.create<String> { sink ->
                                sink.onDispose {
                                    // 会话断开时移除订阅
                                    topicSinks[topicId]?.remove(sessionId)
                                }
                            }
                            session.send(flux.map { TextMessage(it) }).subscribe()
                            flux.sink
                        }
                } else if (isUnsubscribeMessage(payload)) {
                    val topicId = extractTopicIdUnsubscribeMessage(payload)
                    topicSinks[topicId]?.remove(sessionId)
                }
            }

        // 保持连接活跃
        return input.then()
    }

    // 供事件处理器调用的消息发送方法
    fun sendToTopic(topicId: String, message: String) {
        topicSinks[topicId]?.values?.forEach { sink ->
            sink.next(message)
        }
    }
}

// 注册响应式WebSocket路由
@Configuration
class WebfluxWebSocketConfig {
    @Bean
    fun webSocketRouter(handler: ReactiveWebSocketHandler): RouterFunction<ServerResponse> {
        return RouterFunctions.route(
            RequestPredicates.ws("/ws"), handler::handle
        )
    }
}

方案2：混合模式（不推荐） #

若强行保留STOMP Broker能力，需在Webflux应用中引入Spring Web的WebSocket模块，但会引入阻塞式Servlet容器，失去Webflux的响应式优势，得不偿失。

响应式编程与STOMP的架构差异 #

• STOMP是帧请求-响应模型，每个操作（订阅、发送）都是独立的帧交互，属于命令式编程。
• 响应式编程是流处理模型，WebSocket连接是双向流，客户端消息为输入流，服务器消息为输出流，所有操作异步非阻塞，适配高并发场景。

对于你的场景，推荐方案1，既能享受Webflux的高并发优势，又能满足主题订阅的业务需求。

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