带有MongoDB的Spring WebFlux-节流SSE客户端

Question

我正在开发一个简单的聊天服务，该服务由Spring Boot 2.1.1运行，WebFlux、Reactor 3.2.3、Mongo 3.8.2和Netty 4.1.31。

null

观察似乎很明显，因为当我在测试期间通过SSE连接时，当新事件到达时，它几乎立即更新我--基本上，SSE的响应比每2秒轮询高数百倍。

问题是：

假定客户机最终是订阅者（或者至少我认为它是由有限的知识提供的），我能以某种方式抑制ReactiveMongoTemplate发布消息的速度吗？或者以某种方式减少对新事件的需求，而不必在客户端这样做？

// ChatRepository.java

private static final Query chatEventsQuery = new Query();

public Flux<ChatEvent> getChatEventsStream(String chatId) {
    return reactiveMongoTemplate.tail(
            chatEventsQuery,
            ChatEvent.class,
            chatId
    );
}

,

// ChatHandler.java

public Mono<ServerResponse> getChatStream(ServerRequest request) {

    String chatId = request.pathVariable(CHAT_ID_PATH_VARIABLE);
    String username = getUsername(request);

    Flux<ServerSentEvent> chatEventsStream = chatRepository
            .getChatEventsStream(chatId)
            .map(addUserSpecificPropsToChatEvent(username))
            .map(event -> ServerSentEvent.<ChatEvent>builder()
                    .event(event.getType().getEventName())
                    .data(event)
                    .build());

    log.debug("\nExposing chat stream\nchat: {}\nuser: {}", chatId, username);

    return ServerResponse.ok().body(
            chatEventsStream,
            ServerSentEvent.class
    );
}

,

// ChatRouter.java

RouterFunction<ServerResponse> routes(ChatHandler handler) {
    return route(GET("/api/chat/{chatId}/stream"), handler::getChatStream);
}

Answer 1

答案是：您可以使用flux.buffer方法来完成。然后，流量将按定义的速率将事件以块的形式发送给订阅服务器。

我发布的代码有两个主要问题

>

考虑到多个用户通常都在收听一个聊天，我对ChatRepository进行了重构，以利用“热”的、可重放的流量（现在每个聊天有1个流，而不是每个用户有1个流），这些流存储在咖啡因缓存中。另外，我通过短时间间隔缓冲它们，以避免在繁忙的聊天中将事件推送给客户机时占用大量资源。

// ChatRepository.java

public Flux<List<ChatEvent>> getChatEventsStream(String chatId) {
    return Optional.ofNullable(chatStreamsCache.getIfPresent(chatId))
            .orElseGet(newCachedChatEventsStream(chatId))
            .autoConnect();
}

private Supplier<ConnectableFlux<List<ChatEvent>>> newCachedChatEventsStream(String chatId) {
    return () -> {
        ConnectableFlux<List<ChatEvent>> chatEventsStream = reactiveMongoTemplate.tail(
                null,
                ChatEvent.class,
                chatId
        ).buffer(Duration.ofMillis(chatEventsBufferInterval))
                .replay(chatEventsReplayCount);

        chatStreamsCache.put(chatId, chatEventsStream);

        return chatEventsStream;
    };
}

// ChatHandler.java

public Mono<ServerResponse> getChatStream(ServerRequest request) {

    String chatId = request.pathVariable(CHAT_ID_PATH_VARIABLE);
    String username = getUsername(request);

    Flux<ServerSentEvent> chatEventsStream = chatRepository
            .getChatEventsStream(chatId)
            .map(addUserSpecificPropsToChatEvents(username))
            .map(event -> ServerSentEvent.<List<ChatEvent>>builder()
                    .event(CHAT_SSE_NAME)
                    .data(event)
                    .build());

    log.debug("\nExposing chat stream\nchat: {}\nuser: {}", chatId, username);

    return ServerResponse.ok().body(
            chatEventsStream,
            ServerSentEvent.class
    );
}

,

应用这些更改后，即使有3000个活跃用户，该服务也能很好地执行（JVM使用了50%的CPU，Mongo~7%主要是由于大量插入--流现在不那么明显了）