如何确保Spring WebFlux与WebSocket的响应式流正常完成？

我来帮你整理并完善这个Spring WebFlux WebSocket服务端的实现，它的核心功能是接收客户端发来的数字，返回对应数量的递增数字序列～

当WebSocket客户端发送一个数字（比如4），服务端会返回从0到该数字减一的序列（比如0、1、2、3）。下面是完整的实现代码和关键说明：

完整的WebSocket 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

class NumbersWebSocketHandler : WebSocketHandler {

    override fun handle(session: WebSocketSession): Mono<Void> {
        // 先接收客户端发送的数字，解析为需要生成的序列长度
        return session.receive()
            .map { it.payloadAsText.toInt() }
            .flatMap { count ->
                var index = 0
                // 生成从0到count-1的整数序列
                val numberFlux = Flux.generate<Int> { sink ->
                    if (index < count) {
                        sink.next(index++)
                    } else {
                        sink.complete()
                    }
                }
                // 将整数转为字符串，封装成WebSocket消息发送给客户端
                val outputMessages = numberFlux
                    .map(Int::toString)
                    .map(session::textMessage)
                
                // 发送消息并完成会话
                session.send(outputMessages)
            }
            .then()
    }
}

代码关键说明 #

• 会话处理流程：首先通过session.receive()接收客户端的消息，解析为整数类型的序列长度count
• 序列生成：使用Flux.generate逐个生成0到count-1的整数，当index达到count时自动完成序列
• 消息发送：把生成的整数转为字符串，调用session.textMessage()封装成WebSocket消息，再通过session.send()推送给客户端
• 会话完成：最后用.then()确保整个交互流程完成后优雅关闭会话

注册WebSocket Handler到Spring #

要让这个handler能被客户端访问，需要在Spring配置类中注册它：

import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.web.reactive.socket.WebSocketHandlerMapping

@Configuration
class WebSocketConfig {

    @Bean
    fun webSocketHandlerMapping(numbersWebSocketHandler: NumbersWebSocketHandler): WebSocketHandlerMapping {
        val handlerMapping = WebSocketHandlerMapping()
        // 配置handler对应的访问路径
        handlerMapping.urlMap = mapOf("/ws/numbers" to numbersWebSocketHandler)
        handlerMapping.order = 1
        return handlerMapping
    }
}

这样客户端就可以通过ws://localhost:8080/ws/numbers这个地址连接，发送数字后就能收到对应的序列啦～

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