如何为Spring Cloud GatewayFilter配置CircuitBreaker触发降级路由？

我有一个用于获取OAuth2令牌的GatewayFilter，需要接入CircuitBreaker。当无法连接到认证提供商或认证失败时，我需要触发降级路由。

我的过滤器代码如下：

import org.springframework.cloud.gateway.filter.GatewayFilter;
import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.security.oauth2.client.OAuth2AuthorizeRequest;
import org.springframework.security.oauth2.client.ReactiveOAuth2AuthorizedClientManager;
import org.springframework.stereotype.Component;

@Component
public class Oauth2GatewayFilterFactory extends AbstractGatewayFilterFactory<Oauth2GatewayFilterFactory.Config> {

    private final ReactiveOAuth2AuthorizedClientManager authorizedClientManager;

    public Oauth2GatewayFilterFactory(
            ReactiveOAuth2AuthorizedClientManager authorizedClientManager) {
        super(Config.class);
        this.authorizedClientManager = authorizedClientManager;
    }

    @Override
    public GatewayFilter apply(Config config) {
        return (exchange, chain) -> {
            final var clientRegistrationId = config.getClientRegistrationId();
            final var oauth2Request = OAuth2AuthorizeRequest.withClientRegistrationId(clientRegistrationId)
                    .principal("N/A")
                    .build();
            return authorizedClientManager.authorize(oauth2Request)
                    .map(authorizedClient -> exchange.mutate()
                            .request(r -> r.headers(
                                    headers -> headers.set(HttpHeaders.AUTHORIZATION,
                                            "Bearer " + authorizedClient.getAccessToken().getTokenValue())))
                            .build())
                    .defaultIfEmpty(exchange)
                    .flatMap(chain::filter);
        };
    }

    public static class Config {

        private String clientRegistrationId;

        public String getClientRegistrationId() {
            return clientRegistrationId;
        }

        public void setClientRegistrationId(String clientRegistrationId) {
            this.clientRegistrationId = clientRegistrationId;
        }

    }

}

我尝试在过滤器工厂中接入CircuitBreaker并使用transformDeferred：

return authorizedClientManager.authorize(oauth2Request)
        .transformDeferred(CircuitBreakerOperator.of(circuitBreaker))
        .map(authorizedClient -> {

但这种方式没有效果。请问该如何正确接入CircuitBreaker？

要在Reactive GatewayFilter中正确接入CircuitBreaker，需调整流的包裹方式并完善降级逻辑，具体步骤如下：

1. 注入CircuitBreaker实例 #

首先确保项目中引入Resilience4j依赖，并配置CircuitBreaker的Bean（以自定义配置为例）：

import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.time.Duration;

@Configuration
public class CircuitBreakerConfig {

    @Bean
    public CircuitBreaker oauth2TokenCircuitBreaker() {
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
                .failureRateThreshold(50) // 失败率达50%时触发断路
                .waitDurationInOpenState(Duration.ofSeconds(10)) // 断路后10秒尝试恢复
                .permittedNumberOfCallsInHalfOpenState(3) // 半开状态允许3次试探调用
                .slidingWindowSize(10) // 滑动窗口统计最近10次调用
                .build();
        return CircuitBreaker.of("oauth2-token-acquisition", config);
    }
}

2. 修改过滤器代码，正确包裹授权流 #

将CircuitBreaker逻辑融入授权请求的响应流，并添加异常降级处理：

import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.resilience4j.reactor.circuitbreaker.operator.CircuitBreakerOperator;
import org.springframework.cloud.gateway.filter.GatewayFilter;
import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.security.oauth2.client.OAuth2AuthorizeRequest;
import org.springframework.security.oauth2.client.ReactiveOAuth2AuthorizedClientManager;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

@Component
public class Oauth2GatewayFilterFactory extends AbstractGatewayFilterFactory<Oauth2GatewayFilterFactory.Config> {

    private final ReactiveOAuth2AuthorizedClientManager authorizedClientManager;
    private final CircuitBreaker circuitBreaker;

    // 注入CircuitBreaker实例
    public Oauth2GatewayFilterFactory(
            ReactiveOAuth2AuthorizedClientManager authorizedClientManager,
            CircuitBreaker circuitBreaker) {
        super(Config.class);
        this.authorizedClientManager = authorizedClientManager;
        this.circuitBreaker = circuitBreaker;
    }

    @Override
    public GatewayFilter apply(Config config) {
        return (exchange, chain) -> {
            final var clientRegistrationId = config.getClientRegistrationId();
            final var oauth2Request = OAuth2AuthorizeRequest.withClientRegistrationId(clientRegistrationId)
                    .principal("N/A")
                    .build();

            return authorizedClientManager.authorize(oauth2Request)
                    // 用CircuitBreakerOperator包裹整个授权请求流
                    .transform(CircuitBreakerOperator.of(circuitBreaker))
                    .map(authorizedClient -> exchange.mutate()
                            .request(r -> r.headers(
                                    headers -> headers.set(HttpHeaders.AUTHORIZATION,
                                            "Bearer " + authorizedClient.getAccessToken().getTokenValue())))
                            .build())
                    // 处理授权失败或断路触发的异常，执行降级逻辑
                    .onErrorResume(throwable -> {
                        // 方案1：直接返回服务不可用响应
                        exchange.getResponse().setStatusCode(HttpStatus.SERVICE_UNAVAILABLE);
                        return exchange.getResponse().setComplete();
                        // 方案2：转发到降级路由
                        // exchange.getRequest().mutate().path("/oauth-fallback").build();
                        // return chain.filter(exchange);
                    })
                    .defaultIfEmpty(exchange)
                    .flatMap(chain::filter);
        };
    }

    public static class Config {

        private String clientRegistrationId;

        public String getClientRegistrationId() {
            return clientRegistrationId;
        }

        public void setClientRegistrationId(String clientRegistrationId) {
            this.clientRegistrationId = clientRegistrationId;
        }

    }

}

3. 关键说明 #

• 替换transformDeferred为transform：transformDeferred适用于延迟创建操作符的场景，直接使用transform包裹已有授权流更适配当前需求。
• 异常降级逻辑：通过onErrorResume捕获所有授权相关异常（包括CircuitBreaker触发断路的异常），可根据业务需求选择直接返回错误响应或转发到降级服务。
• CircuitBreaker触发条件：自定义配置中的失败率阈值、断路恢复时间等参数，可根据实际认证服务的稳定性调整，确保在服务不可用时快速触发降级，避免请求堆积。

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