Spring Cloud Gateway实现分布式限流和熔断降级的示例代码

        <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 #请求数限流，名字不能乱打
            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 #连接池最大连接数（使用负值表示无极限）
        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;
        //创建线程池
        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 {
                    //访问网关的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 {
        //创建httpclient对象
        closeablehttpclient httpclient = httpclients.createdefault();
        string resultstring = "";
        closeablehttpresponse response = null;
        try {
            //创建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();
            //创建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"));
            //执行请求
            response = httpclient.execute(httpget);
            //判断返回状态是否是200
            if (response.getstatusline().getstatuscode() == 200) {
                resultstring = entityutils.tostring(response.getentity(), "utf-8");
            }
        } catch (exception e) {
            log.info("调用失败:{}",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();
			}
                 // 执行限流
			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);
        // 令牌桶每秒产生令牌数量
		int replenishrate = routeconfig.getreplenishrate();
        // 令牌桶容量
		int burstcapacity = routeconfig.getburstcapacity();
        // 请求消耗的令牌数
		int requestedtokens = routeconfig.getrequestedtokens();
		try {
                  // 键
			list<string> keys = getkeys(id);
                  // 参数
			list<string> scriptargs = arrays.aslist(replenishrate + "", burstcapacity + "", "", requestedtokens + "");
			// 调用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.系统以恒定速率往桶里面放入令牌
-- 2.请求需要被处理，则需要从桶里面获取一个令牌
-- 3.如果桶里面没有令牌可获取，则可以选择等待或直接拒绝并返回
-- 令牌桶算法工作流程
-- 1.计算填满令牌桶所需要的时间（填充时间 = 桶容量 / 速率）
-- 2.设置存储数据的ttl（过期时间），为填充时间的两倍（存储时间 = 填充时间 * 2）
-- 3.从redis获取当前令牌的剩余数量和上一次调用的时间戳
-- 4.计算距离上一次调用的时间间隔（时间间隔 = 当前时间 - 上一次调用时间）
-- 5.计算填充的令牌数量（填充令牌数量 = 时间间隔 * 速率）【前提：桶容量是固定的，不存在无限制的填充】
-- 6.判断是否有足够多的令牌满足请求【 (填充令牌数量 + 剩余令牌数量) >= 请求数量 && (填充令牌数量 + 剩余令牌数量) <= 桶容量 】
-- 7.如果请求被允许，则从桶里面取出相应数据的令牌
-- 8.如果ttl为正，则更新redis键中的令牌和时间戳
-- 9.返回两个两个参数（allowed_num：请求被允许标志。1允许，0不允许）、（new_tokens：填充令牌后剩余的令牌数据）
-- 随机写入
redis.replicate_commands()
-- 令牌桶key -> 存储当前可用令牌的数量（剩余令牌数量）
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])
-- 请求数量
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
-- 获取当前令牌的剩余数量
local last_tokens = tonumber(redis.call("get", tokens_key))
if last_tokens == nil then
    last_tokens = capacity
end
-- 获取上一次调用的时间戳
local last_refreshed = tonumber(redis.call('get', timestamp_key))
if last_refreshed == nil then
    last_refreshed = 0
end
-- 计算距离上一次调用的时间间隔
local delta = math.max(0, now - last_refreshed)
-- 当前的令牌数量（剩余 + 填充 <= 桶容量）
local now_tokens = math.min(capacity, last_refreshed + (rate * delta))
-- 判断是否有足够多的令牌满足请求
local allowed = now_tokens >= requested
-- 定义当前令牌的剩余数量
local new_tokens = now_tokens
-- 定义被允许标志
local allowed_num = 0
if allowed then
    new_tokens = now_tokens - requested
    -- 允许访问
    allowed_num = 1
end
-- ttl > 0，将当前令牌的剩余数量和当前时间戳存入redis
if ttl > 0 then
    redis.call('setex', tokens_key, ttl, new_tokens)
    redis.call('setex', timestamp_key, ttl, now)
end
-- 返回参数
return { allowed_num, new_tokens }

@bean
	@suppresswarnings("unchecked")
	public redisscript redisrequestratelimiterscript() {
		defaultredisscript redisscript = new defaultredisscript<>();
		redisscript.setscriptsource(
                          // 根据指定路径获取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());
    }
}