Spring Cloud Stream SSE供应商仅返回初始消息无后续数据问题排查

我尝试用以下Kotlin代码创建Spring Cloud Stream SSE供应商：

@Bean
fun sseSupplier(sseSupplierPub: Publisher<Message<Any>>): Supplier<Flux<Message<Any>>> {
    return Supplier {
        Flux.from(sseSupplierPub)
    }
}

@Bean
fun sseSupplierPub(
    sseSupplierProperties: sseSupplierProperties
): Publisher<Message<Any>> {
    return IntegrationFlow.from(channelsConfiguration.trigger())
        .enrichHeaders {
            it.header(CONTENT_TYPE, "application/json")
            it.header(ACCEPT, "text/event-stream")
        }
        .handle(sseMessageHandlerSpec())
        .log()
        .toReactivePublisher(true)
}

fun webClient(): WebClient {
    val client = HttpClient.create(provider)
        .keepAlive(true)

    val clientRegistryRepo = InMemoryReactiveClientRegistrationRepository(
        ClientRegistration
            .withRegistrationId("clientId")
            .tokenUri("https://openid-connect/token")
            .clientId("clientId")
            .clientSecret("clientSecret")
            .authorizationGrantType(AuthorizationGrantType.CLIENT_CREDENTIALS)
            .build())
    val clientService = InMemoryReactiveOAuth2AuthorizedClientService(clientRegistryRepo)
    val authorizedClientManager =
        AuthorizedClientServiceReactiveOAuth2AuthorizedClientManager(clientRegistryRepo, clientService)
    val oauthFilter = ServerOAuth2AuthorizedClientExchangeFilterFunction(authorizedClientManager)
    oauthFilter.setDefaultClientRegistrationId("clientId")

    return WebClient.builder()
        .filter(oauthFilter)
        .clientConnector(ReactorClientHttpConnector(client))
        .exchangeStrategies(strategies)
        .build()
}

fun sseMessageHandlerSpec(): WebFluxMessageHandlerSpec {
    return WebFlux.outboundGateway(
        "https://api/src", webClient())
        .httpMethod(HttpMethod.POST)
        .replyPayloadToFlux(true)
        .expectedResponseType(typeReference<ServerSentEvent<String>>())
}

@Bean
fun commandLineRunner(ctx: ApplicationContext?): CommandLineRunner {
    return CommandLineRunner { _ ->
        val message = MessageBuilder.withPayload("{}").build()
        logger.info { "Trigger SSE supplier" }
        channelsConfiguration.trigger().send(message)
    }
}

目前已成功获取授权令牌并连接到https://api/src，但仅收到以下初始消息后无后续推送：

{
    "scanAvailable": true,
    "prefetch": -1
}

可能的原因及排查方向 #

1. SSE响应格式与内容类型不匹配 #

虽然请求头设置了ACCEPT: text/event-stream，但需确认服务端返回的响应头Content-Type是否为text/event-stream。若服务端返回application/json，WebClient会将整个响应解析为单条JSON对象，而非持续的SSE事件流。

可在WebClient中添加日志拦截器验证响应头：

// 在WebClient.builder()中追加
.filter(ExchangeFilterFunction.ofResponseProcessor { response ->
    logger.info("Response Content-Type: ${response.headers().contentType()}")
    Mono.just(response)
})

2. 服务端SSE事件格式不符合标准 #

标准SSE事件需以data:开头，多段数据用换行分隔，最终以\n\n结尾。若服务端返回的初始消息不是标准SSE格式，WebClient可能无法识别后续事件，直接终止流。

用curl直接请求服务端验证格式：

curl -H "Authorization: Bearer YOUR_TOKEN" -H "Accept: text/event-stream" -X POST https://api/src -d "{}"

3. 连接被提前关闭 #

• 服务端超时关闭：部分SSE服务在无事件推送时会超时断连，需确认服务端超时配置，是否需要客户端发送心跳包维持连接。
• 客户端配置缺失：虽设置了keepAlive(true)，但缺少超时参数可能导致连接提前断开，可补充配置：

val client = HttpClient.create(provider)
    .keepAlive(true)
    .responseTimeout(Duration.ofMinutes(30))
    .doOnConnected { conn ->
        conn.addHandlerLast(ReadTimeoutHandler(30, TimeUnit.MINUTES))
    }

4. Integration Flow订阅逻辑问题 #

toReactivePublisher(true)是否正确处理了SSE无限流？若下游订阅是一次性消费，会在收到第一条消息后终止订阅。需确认Supplier<Flux<Message<Any>>>的订阅逻辑是持续监听SSE流，而非仅消费一次。

另外，channelsConfiguration.trigger().send(message)是同步触发，需确认触发后的流是否被持续订阅。

5. 服务端触发条件未满足 #

部分SSE服务需要特定请求参数或后续交互才会推送后续事件。你发送的空{} payload可能仅触发了初始响应，未满足后续推送条件，需核对服务端文档确认请求要求。

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