<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-gateway</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-starter-web</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
    <dependency>
        <groupId>org.apache.httpcomponents</groupId>
        <artifactId>httpclient</artifactId>
    </dependency>

spring:
  application:
    name: laokou-gateway
  cloud:
    gateway:
      routes:
        - id: LAOKOU-SSO-DEMO
          uri: lb://laokou-sso-demo
          predicates:
          - Path=/sso/**
          filters:
          - StripPrefix=1
          - name: RequestRateLimiter #請求數(shù)限流，名字不能亂打
            args:
              key-resolver: "#{@ipKeyResolver}"
              redis-rate-limiter.replenishRate: 1 #生成令牌速率-設為1方便測試
              redis-rate-limiter.burstCapacity: 1 #令牌桶容量-設置1方便測試
  redis:
    database: 0
    cluster:
      nodes: x.x.x.x:7003,x.x.x.x:7004,x.x.x.x:7005,x.x.x.x:7003,x.x.x.x:7004,x.x.x.x:7005
    password: laokou #密碼
    timeout: 6000ms #連接超時時長（毫秒）
    jedis:
      pool:
        max-active: -1 #連接池最大連接數(shù)（使用負值表示無極限）
        max-wait: -1ms #連接池最大阻塞等待時間（使用負值表示沒有限制）
        max-idle: 10 #連接池最大空閑連接
        min-idle: 5 #連接池最小空間連接

@Configuration
public class RequestRateLimiterConfig {

    @Bean(value = "ipKeyResolver")
    public KeyResolver ipKeyResolver(RemoteAddressResolver remoteAddressResolver) {
    	return exchange -> Mono.just(remoteAddressResolver.resolve(exchange).getAddress().getHostAddress());
    }

    @Bean
    public RemoteAddressResolver remoteAddressResolver() {
    	// 遠程地址解析器
    	return XForwardedRemoteAddressResolver.trustAll();
    }

}

@Slf4j
public class HttpUtil {
public static void apiConcurrent(String url,Map<String,String> params) {
        Integer count = 200;
        //創(chuàng)建線程池
        ThreadPoolExecutor pool = new ThreadPoolExecutor(5, 200, 0L, TimeUnit.SECONDS, new SynchronousQueue<>());
        //同步工具
        CountDownLatch latch = new CountDownLatch(count);
        Map<String,String> dataMap = new HashMap<>(1);
        dataMap.put("authorize","XXXXXXX");
        for (int i = 0; i < count; i++) {
            pool.execute(() -> {
                try {
                    //訪問網(wǎng)關的API接口
                    HttpUtil.doGet("http://localhost:1234/sso/laokou-demo/user",dataMap);
                } catch (IOException e) {
                    e.printStackTrace();
                }finally {
                    latch.countDown();
                }
            });
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
public static String doGet(String url, Map<String, String> params) throws IOException {
        //創(chuàng)建HttpClient對象
        CloseableHttpClient httpClient = HttpClients.createDefault();
        String resultString = "";
        CloseableHttpResponse response = null;
        try {
            //創(chuàng)建uri
            URIBuilder builder = new URIBuilder(url);
            if (!params.isEmpty()) {
                for (Map.Entry<String, String> entry : params.entrySet()) {
                    builder.addParameter(entry.getKey(), entry.getValue());
                }
            }
            URI uri = builder.build();
            //創(chuàng)建http GET請求
            HttpGet httpGet = new HttpGet(uri);
            List<NameValuePair> paramList = new ArrayList<>();
            RequestBuilder requestBuilder = RequestBuilder.get().setUri(new URI(url));
            requestBuilder.setEntity(new UrlEncodedFormEntity(paramList, Consts.UTF_8));
            httpGet.setHeader(new BasicHeader("Content-Type", "application/json;charset=UTF-8"));
            httpGet.setHeader(new BasicHeader("Accept", "*/*;charset=utf-8"));
            //執(zhí)行請求
            response = httpClient.execute(httpGet);
            //判斷返回狀態(tài)是否是200
            if (response.getStatusLine().getStatusCode() == 200) {
                resultString = EntityUtils.toString(response.getEntity(), "UTF-8");
            }
        } catch (Exception e) {
            log.info("調(diào)用失敗:{}",e);
        } finally {
            if (response != null) {
                response.close();
            }
            httpClient.close();
        }
        log.info("打?。簕}",resultString);
        return resultString;
    }
}

@Override
	public GatewayFilter apply(Config config) {
		KeyResolver resolver = getOrDefault(config.keyResolver, defaultKeyResolver);
		RateLimiter<Object> limiter = getOrDefault(config.rateLimiter, defaultRateLimiter);
		boolean denyEmpty = getOrDefault(config.denyEmptyKey, this.denyEmptyKey);
		HttpStatusHolder emptyKeyStatus = HttpStatusHolder
				.parse(getOrDefault(config.emptyKeyStatus, this.emptyKeyStatusCode));
		return (exchange, chain) -> resolver.resolve(exchange).defaultIfEmpty(EMPTY_KEY).flatMap(key -> {
			if (EMPTY_KEY.equals(key)) {
				if (denyEmpty) {
					setResponseStatus(exchange, emptyKeyStatus);
					return exchange.getResponse().setComplete();
				}
				return chain.filter(exchange);
			}
			String routeId = config.getRouteId();
			if (routeId == null) {
				Route route = exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR);
				routeId = route.getId();
			}
                 // 執(zhí)行限流
			return limiter.isAllowed(routeId, key).flatMap(response -> {
				for (Map.Entry<String, String> header : response.getHeaders().entrySet()) {
					exchange.getResponse().getHeaders().add(header.getKey(), header.getValue());
				}
				if (response.isAllowed()) {
					return chain.filter(exchange);
				}
				setResponseStatus(exchange, config.getStatusCode());
				return exchange.getResponse().setComplete();
			});
		});
	}

@Override
	@SuppressWarnings("unchecked")
	public Mono<Response> isAllowed(String routeId, String id) {
		if (!this.initialized.get()) {
			throw new IllegalStateException("RedisRateLimiter is not initialized");
		}
        // 這里如何加載配置?請思考
		Config routeConfig = loadConfiguration(routeId);
        // 令牌桶每秒產(chǎn)生令牌數(shù)量
		int replenishRate = routeConfig.getReplenishRate();
        // 令牌桶容量
		int burstCapacity = routeConfig.getBurstCapacity();
        // 請求消耗的令牌數(shù)
		int requestedTokens = routeConfig.getRequestedTokens();
		try {
                  // 鍵
			List<String> keys = getKeys(id);
                  // 參數(shù)
			List<String> scriptArgs = Arrays.asList(replenishRate + "", burstCapacity + "", "", requestedTokens + "");
			// 調(diào)用lua腳本
			Flux<List<Long>> flux = this.redisTemplate.execute(this.script, keys, scriptArgs);
			return flux.onErrorResume(throwable -> {
				if (log.isDebugEnabled()) {
					log.debug("Error calling rate limiter lua", throwable);
				}
				return Flux.just(Arrays.asList(1L, -1L));
			}).reduce(new ArrayList<Long>(), (longs, l) -> {
				longs.addAll(l);
				return longs;
			}).map(results -> {
                          // 判斷是否等于1，1表示允許通過，0表示不允許通過
				boolean allowed = results.get(0) == 1L;
				Long tokensLeft = results.get(1);
				Response response = new Response(allowed, getHeaders(routeConfig, tokensLeft));
				if (log.isDebugEnabled()) {
					log.debug("response: " + response);
				}
				return response;
			});
		}
		catch (Exception e) {
			log.error("Error determining if user allowed from redis", e);
		}
		return Mono.just(new Response(true, getHeaders(routeConfig, -1L)));
	}
	static List<String> getKeys(String id) {
		String prefix = "request_rate_limiter.{" + id;
		String tokenKey = prefix + "}.tokens";
		String timestampKey = prefix + "}.timestamp";
		return Arrays.asList(tokenKey, timestampKey);
	}

-- User Request Rate Limiter filter
-- See https://stripe.com/blog/rate-limiters
-- See https://gist.github.com/ptarjan/e38f45f2dfe601419ca3af937fff574d#file-1-check_request_rate_limiter-rb-L11-L34
-- 令牌桶算法工作原理
-- 1.系統(tǒng)以恒定速率往桶里面放入令牌
-- 2.請求需要被處理，則需要從桶里面獲取一個令牌
-- 3.如果桶里面沒有令牌可獲取，則可以選擇等待或直接拒絕并返回
-- 令牌桶算法工作流程
-- 1.計算填滿令牌桶所需要的時間（填充時間 = 桶容量 / 速率）
-- 2.設置存儲數(shù)據(jù)的TTL（過期時間），為填充時間的兩倍（存儲時間 = 填充時間 * 2）
-- 3.從Redis獲取當前令牌的剩余數(shù)量和上一次調(diào)用的時間戳
-- 4.計算距離上一次調(diào)用的時間間隔（時間間隔 = 當前時間 - 上一次調(diào)用時間）
-- 5.計算填充的令牌數(shù)量（填充令牌數(shù)量 = 時間間隔 * 速率）【前提：桶容量是固定的，不存在無限制的填充】
-- 6.判斷是否有足夠多的令牌滿足請求【 (填充令牌數(shù)量 + 剩余令牌數(shù)量) >= 請求數(shù)量 && (填充令牌數(shù)量 + 剩余令牌數(shù)量) <= 桶容量 】
-- 7.如果請求被允許，則從桶里面取出相應數(shù)據(jù)的令牌
-- 8.如果TTL為正，則更新Redis鍵中的令牌和時間戳
-- 9.返回兩個兩個參數(shù)（allowed_num：請求被允許標志。1允許，0不允許）、（new_tokens：填充令牌后剩余的令牌數(shù)據(jù)）
-- 隨機寫入
redis.replicate_commands()
-- 令牌桶Key -> 存儲當前可用令牌的數(shù)量（剩余令牌數(shù)量）
local tokens_key = KEYS[1]
-- 時間戳Key -> 存儲上次令牌刷新的時間戳
local timestamp_key = KEYS[2]
-- 令牌填充速率
local rate = tonumber(ARGV[1])
-- 令牌桶容量
local capacity = tonumber(ARGV[2])
-- 當前時間
local now = tonumber(ARGV[3])
-- 請求數(shù)量
local requested = tonumber(ARGV[4])
-- 填滿令牌桶所需要的時間
local fill_time = capacity / rate
-- 設置key的過期時間（填滿令牌桶所需時間的2倍）
local ttl = math.floor(fill_time * 2)
-- 判斷當前時間，為空則從redis獲取
if now == nil then
    now = redis.call('TIME')[1]
end
-- 獲取當前令牌的剩余數(shù)量
local last_tokens = tonumber(redis.call("get", tokens_key))
if last_tokens == nil then
    last_tokens = capacity
end
-- 獲取上一次調(diào)用的時間戳
local last_refreshed = tonumber(redis.call('get', timestamp_key))
if last_refreshed == nil then
    last_refreshed = 0
end
-- 計算距離上一次調(diào)用的時間間隔
local delta = math.max(0, now - last_refreshed)
-- 當前的令牌數(shù)量（剩余 + 填充 <= 桶容量）
local now_tokens = math.min(capacity, last_refreshed + (rate * delta))
-- 判斷是否有足夠多的令牌滿足請求
local allowed = now_tokens >= requested
-- 定義當前令牌的剩余數(shù)量
local new_tokens = now_tokens
-- 定義被允許標志
local allowed_num = 0
if allowed then
    new_tokens = now_tokens - requested
    -- 允許訪問
    allowed_num = 1
end
-- ttl > 0，將當前令牌的剩余數(shù)量和當前時間戳存入redis
if ttl > 0 then
    redis.call('setex', tokens_key, ttl, new_tokens)
    redis.call('setex', timestamp_key, ttl, now)
end
-- 返回參數(shù)
return { allowed_num, new_tokens }

@Bean
	@SuppressWarnings("unchecked")
	public RedisScript redisRequestRateLimiterScript() {
		DefaultRedisScript redisScript = new DefaultRedisScript<>();
		redisScript.setScriptSource(
                          // 根據(jù)指定路徑獲取lua腳本來初始化配置
				new ResourceScriptSource(new ClassPathResource("META-INF/scripts/request_rate_limiter.lua")));
		redisScript.setResultType(List.class);
		return redisScript;
	}
	@Bean
	@ConditionalOnMissingBean
	public RedisRateLimiter redisRateLimiter(ReactiveStringRedisTemplate redisTemplate,
			@Qualifier(RedisRateLimiter.REDIS_SCRIPT_NAME) RedisScript<List<Long>> redisScript,
			ConfigurationService configurationService) {
		return new RedisRateLimiter(redisTemplate, redisScript, configurationService);
	}

spring:
  cloud:
    gateway:
      server:
        webflux:
          filter:
            request-rate-limiter:
              enabled: true

@Bean
@ConditionalOnBean({ RateLimiter.class, KeyResolver.class })
@ConditionalOnEnabledFilter
public RequestRateLimiterGatewayFilterFactory requestRateLimiterGatewayFilterFactory(RateLimiter rateLimiter,
       KeyResolver resolver) {
    return new RequestRateLimiterGatewayFilterFactory(rateLimiter, resolver);
}

@Retention(RetentionPolicy.RUNTIME)
@Target({ ElementType.TYPE, ElementType.METHOD })
@Documented
@Conditional(OnEnabledFilter.class)
public @interface ConditionalOnEnabledFilter {
    // 這里value是用來指定滿足條件的某些類，換一句話說，就是這些類都加載或注入到ioc容器，這個注解修飾的自動裝配類才會生效
    Class<? extends GatewayFilterFactory<?>> value() default OnEnabledFilter.DefaultValue.class;
}

public class OnEnabledFilter extends OnEnabledComponent<GatewayFilterFactory<?>> {}

public abstract class OnEnabledComponent<T> extends SpringBootCondition implements ConfigurationCondition {
    private static final String PREFIX = "spring.cloud.gateway.server.webflux.";
    private static final String SUFFIX = ".enabled";
    private ConditionOutcome determineOutcome(Class<? extends T> componentClass, PropertyResolver resolver) {
       // 拼接完整名稱
       // 例如 => spring.cloud.gateway.server.webflux.request-rate-limiter.enabled
       String key = PREFIX + normalizeComponentName(componentClass) + SUFFIX;
       ConditionMessage.Builder messageBuilder = forCondition(annotationClass().getName(), componentClass.getName());
       if ("false".equalsIgnoreCase(resolver.getProperty(key))) {
          // 不滿足條件不加載bean
          return ConditionOutcome.noMatch(messageBuilder.because("bean is not available"));
       }
       // 滿足條件加載bean
       return ConditionOutcome.match();
    }
}

@Getter
@ConfigurationProperties("spring.cloud.gateway.server.webflux.filter.request-rate-limiter")
public class RequestRateLimiterGatewayFilterFactory
       extends AbstractGatewayFilterFactory<RequestRateLimiterGatewayFilterFactory.Config> {
    private static final String EMPTY_KEY = "____EMPTY_KEY__";
    private final RateLimiter<?> defaultRateLimiter;
    private final KeyResolver defaultKeyResolver;
    /**
     * Switch to deny requests if the Key Resolver returns an empty key, defaults to true.
     */
    @Setter
    private boolean denyEmptyKey = true;
    /** HttpStatus to return when denyEmptyKey is true, defaults to FORBIDDEN. */
    @Setter
    private String emptyKeyStatusCode = HttpStatus.FORBIDDEN.name();
    public RequestRateLimiterGatewayFilterFactory(RateLimiter<?> defaultRateLimiter, KeyResolver defaultKeyResolver) {
       super(Config.class);
       this.defaultRateLimiter = defaultRateLimiter;
       this.defaultKeyResolver = defaultKeyResolver;
    }
    @Override
    public GatewayFilter apply(Config config) {
       KeyResolver resolver = getOrDefault(config.keyResolver, defaultKeyResolver);
       RateLimiter<?> limiter = getOrDefault(config.rateLimiter, defaultRateLimiter);
       boolean denyEmpty = getOrDefault(config.denyEmptyKey, this.denyEmptyKey);
       HttpStatusHolder emptyKeyStatus = HttpStatusHolder
          .parse(getOrDefault(config.emptyKeyStatus, this.emptyKeyStatusCode));
       return (exchange, chain) -> resolver.resolve(exchange).defaultIfEmpty(EMPTY_KEY).flatMap(key -> {
          if (EMPTY_KEY.equals(key)) {
             if (denyEmpty) {
                setResponseStatus(exchange, emptyKeyStatus);
                return exchange.getResponse().setComplete();
             }
             return chain.filter(exchange);
          }
          String routeId = config.getRouteId();
          if (routeId == null) {
             Route route = exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR);
             Assert.notNull(route, "Route is null");
             routeId = route.getId();
          }
          return limiter.isAllowed(routeId, key).flatMap(response -> {
             for (Map.Entry<String, String> header : response.getHeaders().entrySet()) {
                exchange.getResponse().getHeaders().add(header.getKey(), header.getValue());
             }
             if (response.isAllowed()) {
                return chain.filter(exchange);
             }
             // 主要修改這行
             return responseOk(exchange, Result.fail("Too_Many_Requests", "請求太頻繁"));
          });
       });
    }
    private Mono<Void> responseOk(ServerWebExchange exchange, Object data) {
        return responseOk(exchange, JacksonUtils.toJsonStr(data), MediaType.APPLICATION_JSON);
    }
    private Mono<Void> responseOk(ServerWebExchange exchange, String str, MediaType contentType) {
        DataBuffer buffer = exchange.getResponse().bufferFactory().wrap(str.getBytes(StandardCharsets.UTF_8));
        ServerHttpResponse response = exchange.getResponse();
        response.setStatusCode(HttpStatus.OK);
        response.getHeaders().setContentType(contentType);
        response.getHeaders().setContentLength(str.getBytes(StandardCharsets.UTF_8).length);
        return response.writeWith(Flux.just(buffer));
    }
    private <T> T getOrDefault(T configValue, T defaultValue) {
       return (configValue != null) ? configValue : defaultValue;
    }
    public static class Config implements HasRouteId {
       @Getter
       private KeyResolver keyResolver;
       @Getter
       private RateLimiter<?> rateLimiter;
       @Getter
       private HttpStatus statusCode = HttpStatus.TOO_MANY_REQUESTS;
       @Getter
       private Boolean denyEmptyKey;
       @Getter
       private String emptyKeyStatus;
       private String routeId;
       public Config setKeyResolver(KeyResolver keyResolver) {
          this.keyResolver = keyResolver;
          return this;
       }
       public Config setRateLimiter(RateLimiter<?> rateLimiter) {
          this.rateLimiter = rateLimiter;
          return this;
       }
       public Config setStatusCode(HttpStatus statusCode) {
          this.statusCode = statusCode;
          return this;
       }
       public Config setDenyEmptyKey(Boolean denyEmptyKey) {
          this.denyEmptyKey = denyEmptyKey;
          return this;
       }
       public Config setEmptyKeyStatus(String emptyKeyStatus) {
          this.emptyKeyStatus = emptyKeyStatus;
          return this;
       }
       @Override
       public void setRouteId(String routeId) {
          this.routeId = routeId;
       }
       @Override
       public String getRouteId() {
          return this.routeId;
       }
    }
}

<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-starter-circuitbreaker-reactor-resilience4j</artifactId>
</dependency>

spring:
  application:
    name: laokou-gateway
  cloud:
    gateway:
       server:
        webflux:
          routes:
            - id: LAOKOU-SSO-DEMO
              uri: lb://laokou-sso-demo
              predicates:
              - Path=/sso/**
              filters:
              - name: CircuitBreaker
                args:
                  name: default
                  fallbackUri: "forward:/fallback"
          filter:
            circuit-breaker:
              enabled: true

/**
 * @author laokou
 */
@Configuration
public class CircuitBreakerConfig {
    @Bean
    public RouterFunction<ServerResponse> routerFunction() {
       return RouterFunctions.route(
             RequestPredicates.path("/fallback").and(RequestPredicates.accept(MediaType.TEXT_PLAIN)),
             (request) -> ServerResponse.status(HttpStatus.SC_OK)
                .contentType(MediaType.APPLICATION_JSON)
                .body(BodyInserters.fromValue(Result.fail("Service_Unavailable", "服務正在維護"))));
    }
    @Bean
    public Customizer<ReactiveResilience4JCircuitBreakerFactory> reactiveResilience4JCircuitBreakerFactoryCustomizer() {
       return factory -> factory.configureDefault(id -> new Resilience4JConfigBuilder(id)
          // 3秒后超時時間
          .timeLimiterConfig(TimeLimiterConfig.custom().timeoutDuration(Duration.ofSeconds(3)).build())
          .circuitBreakerConfig(io.github.resilience4j.circuitbreaker.CircuitBreakerConfig.ofDefaults())
          .build());
    }
}