Spring WebFlux SSE重试时如何更新Last-Event-ID请求头

问题根源在于原代码中请求对象req是一次性初始化的，重试时复用的是同一个已设置好初始Last-Event-ID的实例，无法动态更新头信息。要解决这个问题，需要让每次重试时都重新构建请求，加载最新的Last-Event-ID。

具体实现方案 #

使用Mono.defer()包装请求构建逻辑，defer会在每次订阅（包括重试触发的重新订阅）时执行内部代码，这样每次都能获取最新的事件ID并设置到请求头中。

修改后的完整代码 #

var type = new ParameterizedTypeReference<ServerSentEvent<String>>() {};

Mono.defer(() -> {
        // 每次订阅（含重试）都会重新加载最新的Last-Event-ID
        String lastEventId = loadLastId();
        var req = webClient.get()
                .uri("/the/end/point");
        
        if (lastEventId != null && !lastEventId.isEmpty()) {
            req = req.header("Last-Event-ID", lastEventId);
        }
        
        return req.retrieve().bodyToFlux(type);
    })
    .retryWhen(Retry
        .backoff(maxRetries, Duration.ofSeconds(retryBackoffStep))
        .maxBackoff(Duration.ofSeconds(10))
        .transientErrors(true)
        .filter(t -> {
            log.warn("SSE Notification Channel: connection error detected", t);
            return !(t instanceof ClientAuthorizationException)
                && !(t instanceof WebClientResponseException.Forbidden)
                && !(t instanceof WebClientResponseException.NotFound);
        })
        .doBeforeRetry(signal -> {
            log.warn("SSE Notification Channel: retry SSE connection, attempt {}",
                            signal.totalRetriesInARow() + 1);
        })
        .doAfterRetry(signal ->
            log.warn("SSE Notification Channel: retried SSE connection, attempt {}, successful = {}",
                            signal.totalRetriesInARow() + 1, signal.failure() == null))
        .onRetryExhaustedThrow((retryBackoffSpec, signal) -> {
            log.error("SSE Notification Channel: SSE connection retries exhausted");
            return signal.failure();
        })
    )
    .subscribe(
        this::handleContent,
        this::handleError,
        () -> log.error("SSE Notification Channel: SSE connection closed")
    );

关键改动说明 #

• 将请求构建逻辑移入Mono.defer()内部，确保每次重试触发重新订阅时，都会执行loadLastId()获取最新的事件ID，再动态设置请求头。
• 替换原有的一次性请求对象为动态生成的请求实例，从根本上解决重试时头信息无法更新的问题。

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