Spring Cloud Gateway迁移：响应式范式下实现AuthToken单次获取

问题根源 #

命令式代码里用synchronized能保证同一时间只有一个线程去调用第三方API，但响应式范式下不能用阻塞同步——这会破坏非阻塞线程模型。而单纯用Mono.cache()无法解决并发问题：当多个请求同时订阅还未完成的Mono时，每个订阅都会触发上游的第三方API调用，导致重复请求。

解决方案：用原子引用控制Token请求的临界区 #

核心思路是用AtomicReference原子性地持有当前有效的Token请求Mono，确保并发请求都订阅同一个Mono实例，避免重复调用第三方API。同时结合Mono.cache()实现Token的过期与闲置超时刷新。

1. 实现Token提供者 #

import reactor.core.publisher.Mono;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicReference;

public class AuthTokenProvider {
    private final AtomicReference<Mono<String>> currentTokenMono = new AtomicReference<>();

    // 实际调用第三方API获取Token的响应式方法
    private Mono<String> fetchNewToken() {
        // 替换为你的WebClient调用逻辑，比如：
        // return webClient.post()
        //         .uri("/auth/token")
        //         .bodyValue(authRequest)
        //         .retrieve()
        //         .bodyToMono(String.class);
        
        // 模拟第三方API调用（仅作示例）
        return Mono.fromCallable(() -> {
            System.out.println("Calling third-party auth API...");
            return "Bearer " + System.currentTimeMillis();
        });
    }

    public Mono<String> getValidToken() {
        Mono<String> existingMono = currentTokenMono.get();
        // 检查是否存在有效的未过期Token请求
        if (existingMono == null || existingMono.isDisposed()) {
            // 创建新的Token请求Mono，配置缓存规则
            Mono<String> newTokenMono = fetchNewToken()
                    .cache(
                        // Token过期时间：比如1小时后自动失效
                        token -> Mono.delay(Duration.ofHours(1)),
                        // 闲置超时：30分钟无请求则刷新Token
                        Mono.delay(Duration.ofMinutes(30))
                    )
                    // 请求完成/取消后清除原子引用，确保下次请求重新获取
                    .doFinally(signalType -> currentTokenMono.compareAndSet(newTokenMono, null));
            
            // CAS原子操作，确保只有一个线程能成功设置新的Mono
            if (currentTokenMono.compareAndSet(existingMono, newTokenMono)) {
                existingMono = newTokenMono;
            } else {
                // 其他线程已创建新Mono，直接使用已存在的实例
                existingMono = currentTokenMono.get();
            }
        }
        return existingMono;
    }
}

2. 集成到Spring Cloud Gateway过滤器 #

import org.springframework.cloud.gateway.filter.GatewayFilter;
import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory;
import reactor.core.publisher.Mono;

public class GlobalAuthTokenFilter extends AbstractGatewayFilterFactory<GlobalAuthTokenFilter.Config> {
    private final AuthTokenProvider tokenProvider;

    public GlobalAuthTokenFilter(AuthTokenProvider tokenProvider) {
        this.tokenProvider = tokenProvider;
        super(Config.class);
    }

    @Override
    public GatewayFilter apply(Config config) {
        return (exchange, chain) -> tokenProvider.getValidToken()
                .map(token -> {
                    // 将Token注入请求头
                    exchange.getRequest().mutate().header("Authorization", token);
                    return exchange;
                })
                .flatMap(chain::filter);
    }

    // 过滤器配置类（可根据需求添加参数）
    public static class Config {}
}

3. 注册过滤器为全局过滤器 #

在配置类中注册：

import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class GatewayConfig {
    @Bean
    public GlobalFilter globalAuthTokenFilter(AuthTokenProvider tokenProvider) {
        return new GlobalAuthTokenFilter(tokenProvider);
    }

    @Bean
    public AuthTokenProvider authTokenProvider() {
        return new AuthTokenProvider();
    }
}

关键细节说明 #

• 原子引用+CAS操作：保证并发场景下只有一个Mono被创建去调用第三方API，避免重复请求。
• Mono.cache()的超时配置：第一个参数控制Token的绝对过期时间，第二个参数控制闲置超时——当Token超过指定时间没有被订阅时，自动失效并触发重新获取。
• 非阻塞设计：全程不使用阻塞同步锁，完全符合响应式范式，不会阻塞Spring Cloud Gateway的线程池。
• 失败处理：可以在fetchNewToken()中添加retryWhen()实现重试逻辑，避免单次API调用失败导致所有请求受阻。

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