使用WebFlux消费事件流时如何提取响应头？

你不需要嵌套subscribe，可以通过Reactor的操作符链式调用实现，既提取响应头，又正常消费事件流。核心思路是利用flatMapMany展开响应实体中的Flux，同时捕获响应头的值供后续使用。

优化后的代码示例 #

webClient.get()
    .uri("/events")
    .accept(MediaType.TEXT_EVENT_STREAM)
    .retrieve()
    .toEntityFlux(EventStreamBatch::class.java)
    .flatMapMany { entity ->
        // 一次性提取响应头（SSE连接建立时返回，全程不变）
        val headerValue = entity.headers["my-header"]?.firstOrNull()
        
        // 展开body的Flux，同时将每个事件批次与响应头关联
        entity.body ?: Flux.empty()
            .map { batch -> batch to headerValue }
    }
    .retry()
    .subscribe(
        { (batch, header) ->
            // 消费事件批次，同时可使用响应头
            eventsConsumer.accept(batch.events.map { it.toInternalEvent() })
            // 示例：打印响应头（根据业务需求调整）
            header?.let { log.info("当前批次关联的头信息：$it") }
        },
        { error ->
            log.error("消费事件流出错", error)
        }
    )

另一种简化写法（直接在流中处理） #

如果不需要将头信息与每个批次绑定，也可以在flatMapMany内部直接处理事件，同时复用头信息：

webClient.get()
    .uri("/events")
    .accept(MediaType.TEXT_EVENT_STREAM)
    .retrieve()
    .toEntityFlux(EventStreamBatch::class.java)
    .flatMapMany { entity ->
        val headerValue = entity.headers["my-header"]?.firstOrNull()
        
        entity.body ?: Flux.empty()
            .doOnNext { batch ->
                // 直接消费事件，同时使用头信息
                eventsConsumer.accept(batch.events.map { it.toInternalEvent() })
                // 头信息相关逻辑
                if (headerValue != null) {
                    // 比如验证头信息、记录日志等
                }
            }
    }
    .retry()
    .subscribe(
        {}, // 已在doOnNext中处理消费，此处可留空或做额外收尾
        { error ->
            log.error("消费事件流出错", error)
        }
    )

关键说明 #

1. 避免嵌套subscribe：嵌套订阅会破坏Reactor的背压机制和生命周期管理，链式操作符能保证流的一致性和错误处理的连贯性。
2. 响应头只提取一次：SSE的响应头在连接建立时就已返回，全程不会变化，所以在flatMapMany中一次性提取即可，无需重复获取。
3. 处理body为空的情况：通过entity.body ?: Flux.empty()避免空指针异常，保证流的稳定性。

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