如何提升Spring WebFlux中WebSocketSession的消息数量限制

我基于Spring WebFlux（Reactor）开发了一个简易聊天服务，其WebSocketHandler代码如下：

public class ChatWebSocketHandler implements WebSocketHandler {
    private final Flux<String> outputEvents;
    private final Sinks.Many<Message> eventPublisher = Sinks.many().unicast().onBackpressureBuffer();

    public ChatWebSocketHandler(){
        this.outputEvents = Flux.from(messages)
                .replay(10)
                .autoConnect();
    }

    @Override
    public Mono<Void> handle(final WebSocketSession session) {
        return session.receive()
                .log()
                .map(WebSocketMessage::getPayloadAsText)
                .filter(msg -> messageValidationService.isMessageValid(msg.getText()))
                .doOnNext(msg -> eventPublisher.tryEmitNext(msg))
                .zipWith(session.send(outputEvents.map(session::textMessage)))
                .then();
    }
}

该服务仅能正常处理前32条消息，消息收发均符合预期，但超过32条后服务器停止响应，无任何日志输出（即使是debug级别）。客户端重新连接后可再次发送32条消息。

.log()的日志如下：

2023-10-04T14:33:37.317+03:00  INFO 15812 --- [ctor-http-nio-2] reactor.Flux.Peek.1                      : onSubscribe(FluxPeek.PeekSubscriber)
2023-10-04T14:33:37.320+03:00  INFO 15812 --- [ctor-http-nio-2] reactor.Flux.Peek.1                      : request(32)
2023-10-04T14:33:37.383+03:00  INFO 15812 --- [ctor-http-nio-2] reactor.Flux.Peek.1                      : onNext("text")
2023-10-04T14:33:37.383+03:00  INFO 15812 --- [ctor-http-nio-2] reactor.Flux.Peek.1                      : onNext("text")
2023-10-04T14:33:37.383+03:00  INFO 15812 --- [ctor-http-nio-2] reactor.Flux.Peek.1                      : onNext("text")
... another 29 same lines

请问该限制设置在何处？是否可以提升该限制？

问题原因与解决方案 #

1. 限制的来源 #

这个32的限制是Reactor Netty WebSocket默认的预取（prefetch）缓冲区大小。从日志里的request(32)可以看出，Reactor Netty在建立WebSocket连接时，会默认向上游请求32条消息的配额。当这32条消息被消费完毕后，如果代码逻辑没有正确触发新的请求（背压机制未正常传播），服务器就会停止接收后续消息。

你的代码中使用zipWith将接收流与发送流合并，zipWith要求两个流的元素一一对应才能继续处理。但聊天场景下，接收消息和广播消息是独立的动作，这种绑定方式会导致接收流的请求传播被阻塞，耗尽初始32个请求配额后，没有新的请求被触发，最终服务器停止响应。

2. 解决与优化方案 #

方案一：调整WebSocket预取大小

可以通过配置Reactor Netty的WebSocket参数修改初始请求量。在Spring Boot中，自定义HttpServer配置即可：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import reactor.netty.http.server.HttpServer;

@Configuration
public class WebSocketConfig {
    @Bean
    public HttpServer httpServer() {
        return HttpServer.create()
                .websocketConfiguration(ws -> ws.prefetch(128)); // 设置自定义预取数量，如128
    }
}

方案二：修复代码中的背压逻辑

核心问题是zipWith的误用，应将接收流与发送流独立处理，避免一一绑定导致的背压阻塞。修改handle方法：

@Override
public Mono<Void> handle(final WebSocketSession session) {
    // 处理接收消息，转发到事件发布器
    Mono<Void> receiveTask = session.receive()
            .log()
            .map(WebSocketMessage::getPayloadAsText)
            .filter(messageValidationService::isMessageValid)
            .doOnNext(eventPublisher::tryEmitNext)
            .then();

    // 订阅事件流，向客户端发送消息
    Mono<Void> sendTask = session.send(outputEvents.map(session::textMessage));

    // 合并两个任务，直到连接关闭
    return Mono.zip(receiveTask, sendTask).then();
}

修改后，接收流与发送流各自独立处理，背压会正常传播，不会再出现请求配额耗尽的问题。若仍需调整预取大小，可结合方案一的配置。

3. 额外说明 #

• 原代码构造函数中的Flux.from(messages)应为Flux.from(eventPublisher)，属于笔误，否则代码无法编译。
• 当前使用Sinks.many().unicast()仅支持单订阅者，若要实现多客户端广播，应替换为Sinks.many().multicast().onBackpressureBuffer()。

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