likes
comments
collection
share

「微服务网关实战四」随意扩展定制的分布式限流，看看我怎么做

作者站长头像
站长
· 阅读数 26

又来给大家更文了。

上一篇讲解过 SCG 的许多基础概念之后，本篇带大家来实战一下网关限流，以及针对网关限流做一个简单的源码分析，网关限流的核心类就是 SCG 中的一个限流过滤器。

限流这件事，你可以在网关做也可以在业务服务做，这个取决于自己的需求，我原来的业务是在业务服务做的限流，后来我转移到了网关里面，因为业务服务内做限流会让业务服务处理很多不该处理的请求，这会在一定程度上阻碍其他正常的业务请求，而网关的吞吐量又比较大，所以我最终在网关层做了限流。

除外以上这一点以外，还要考虑一下转移成本，SCG 中的限流是用网关过滤器做的，而网关过滤器又是配合路由使用的，这代表着你只需要使用 SCG 加几下配置就可以针对某个服务做限流（路由），也可以针对所有服务做限流 （默认过滤器），所以如果是转移限流到 SCG 中，那转移成本将会极低，还带来了更大的扩展性。

接下来就开始今天的正文吧~

1. 限流配置

在 SCG 中，处理限流的网关过滤器是：RequestRateLimiter，在源码中它的类名叫做：RequestRateLimiterGatewayFilterFactory。

由于网关的限流是一个分布式限流，所以必须使用一个中心式的存储来存储限流的状态，在 SCG 使用的就是 Redis，所以你在使用 SCG 的限流器之间首先要引入一个 Redis 依赖，而且由于 SCG 整体是 Reactor 模式的设计，所以你的 Redis 依赖也必须是响应式的：

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
        </dependency>

引入这个依赖之后，你需要在配置文件中配置上 Redis 的地址：

spring:
  redis:
    host: 172.31.128.158
    database: 0
    port: 6379
    connect-timeout: 5000
    password: root

配置地址完成后你可以启动一下服务查看 Redis 是否能成功连接，没有报错就证明你已经连接成功了。

接下来就该到我们的路由配置环节了，依然使用之前的 user-api 的路由配置为例，我们先配置一个 Path 断言，紧接着给它配置一个过滤器：RequestRateLimiter：

    @Bean
    public RouteLocator routeLocator(RouteLocatorBuilder builder) {
        return builder.routes()
                .route((r) -> r.path("/user/**")
                        .filters(f -> f.requestRateLimiter()
                                .rateLimiter(RedisRateLimiter.class, c -> c.setReplenishRate(1).setBurstCapacity(10).setRequestedTokens(5))
                                .configure(c -> c.setKeyResolver(apiTokenKeyResolver()).setDenyEmptyKey(true)))
                        .uri("lb://user-api"))
                .build();
    }

配置的过程中需要先选择先调用 requestRateLimiter() 方法创建一个过滤器实例，然后采用两个方法配置上我们需要的参数，第一个 rateLimiter 方法是配置限流相关的参数，第二个 configure方法则是和限流器相辅助的一些参数：

「微服务网关实战四」随意扩展定制的分布式限流，看看我怎么做

通过上面的示例可以看到我们主要应该配置三个参数（第四个路由参数可以忽略）：

  1. denyEmptyKey：它代表是否允许空 key 通过。
  2. keyResolver：它代表限流计算策略。
  3. rateLimiter：它代表限流算法配置。

通过我这两句干巴巴的解释你可能对这三个配置还是一脸懵的状态，不要紧，我们慢慢来，从下到上逐次开始解释。

2. 限流算法配置

我们先从限流算法的配置开始说起，也就是 c -> c.setReplenishRate(1).setBurstCapacity(10).setRequestedTokens(5) 这一行代码，通过 SCG 的官网我们可以得知 SCG 使用的限流算法是令牌桶算法，计算机领域限流算法有很多，而令牌桶算法则是其中比较完备的算法，它的主要原理是这样的：

「微服务网关实战四」随意扩展定制的分布式限流，看看我怎么做

这是我在网上找到的一张比较形象生动的关于令牌桶工作流程介绍的图（如有侵权，立即删除）。

在令牌桶中，会以一个固定的速率不断的向桶中发送令牌，比如一秒一个，同时这个桶会有一个最大值比如10个，如果发送令牌时发现桶已经达到了最大值，则令牌丢弃。

当有请求到达时，会去检查桶中是否有令牌，如果令牌数满足条件则请求放行，放行之后将令牌数减掉放行的条件，比如每经过一个请求就减掉一个令牌。

那么通过我上面这段描述你可以发现，我举得例子所达到的效果是一个1秒一次的过滤器，因为每秒会增加一个令牌，一个请求又会消耗掉一个令牌，这样在此限流器的作用下就会形成一秒最多经过一个请求的限制。

除此之外，我们还有一个桶最大值的这么一个条件，它代表了我们可以接收一定的突发流量，还是刚才的配置，比如我们的桶已满，这时在一秒内同时到达十个请求，是也满足放行条件的，因为我们的桶里是具有十个令牌的。

当然如果你不想要突发流量这种处理，可以把最大值设置的和你每次请求的消耗值一样，这样请求速率就是恒定的了。

OK，刚刚我简单描述了一下令牌桶的工作原理，那么接下来我们来看一下 SCG 这里配置的实际代码是怎样的，我将本节开头调用的方法的定义贴在这里：

        public <C, R extends RateLimiter<C>> RequestRateLimiterSpec rateLimiter(Class<R> rateLimiterType,
                Consumer<C> configConsumer) {
            R rateLimiter = getBean(rateLimiterType);
            C config = rateLimiter.newConfig();
            configConsumer.accept(config);
            rateLimiter.getConfig().put(routeBuilder.getId(), config);
            return this;
        }

这个方法需要传入一个泛型参数和一个配置类，然后返回一个 RateLimiter 类型的返回值，我点进这个类型一看发现它是一个接口，同时这个接口还继承了 Configurable 接口，通过继承关系我们可以知道 RateLimiter 这个接口它也是一个配置类。

有了接口之后我就去找它的实现类，通过 IDEA 可以看出它只有一个实现类：RedisRateLimiter：

「微服务网关实战四」随意扩展定制的分布式限流，看看我怎么做

那么我们肯定只能去使用这个类了，点进这个类的构造方法你可以看到其中一个构造方法是这样的：

    public RedisRateLimiter(int defaultReplenishRate, int defaultBurstCapacity, int defaultRequestedTokens) {
        this(defaultReplenishRate, defaultBurstCapacity);
        this.defaultConfig.setRequestedTokens(defaultRequestedTokens);
    }

我们跟着它的参数名字一翻译，这不就正对我刚刚讲解令牌桶时的三个概念吗？

  1. 令牌生成速率
  2. 令牌桶最大值
  3. 请求消耗令牌数

把它们和我前面举得 c -> c.setReplenishRate(1).setBurstCapacity(10).setRequestedTokens(5) 对应一下，你是不是突然就知道我的配置什么意思了？

每秒生成一个令牌，但是一次请求需要消耗五个令牌，这代表速率限制为五秒一次，同时配置了令牌桶最大为10，这代表可以允许最大两个请求的突发流量。

我在好早之前开始看 SCG 限流的一些博客的时候，有一个博客也在讲这个，它有一段话我记忆尤新，它的大概意思如下：

defaultReplenishRate 设置的是每秒的速率所以限流最多做到秒级别，无法做到分钟级别。

后来我又翻到了一些博客，我发现许多博客讲限流的时候只说了两个参数：defaultReplenishRate 和 defaultBurstCapacity，并没有说第三个参数：defaultRequestedTokens。

这时我才明白，很多人可能学习的时候压根没有翻阅源码甚至没看过官方文档（官方文档有三个参数示例），他们并不知道有第三个参数，所以才认为限流只能做到秒级别，不知道其实自带限流是可以做到分钟 / 小时级别的。

3. 限流key配置

.setKeyResolver(apiTokenKeyResolver()) 这一行代码的作用是配置一个限流Key规则的对象。

什么叫限流 Key 呢？其实就是你打算用什么当限流标识，比如你在做一个手机验证码的业务，你肯定是限制这个手机号一分钟只能发一次或者30秒只能发一次，这时手机号就是一个限流标识。

而我用的示例则是一个防刷接口，那么我就限制一个用户一分钟最多请求一个接口60次这样子，这时用户ID+请求接口路径是一个限流标识。

我们接着来说回上面那句代码，先来看一下方法定义：

        public Config setKeyResolver(KeyResolver keyResolver) {
            this.keyResolver = keyResolver;
            return this;
        }

这个 KeyResolver 是一个接口，我们只需要写一个此接口的实现类即可：

// 接口定义
public interface KeyResolver {
​
    Mono<String> resolve(ServerWebExchange exchange);
​
}
// 示例
    KeyResolver apiTokenKeyResolver() {
        return exchange -> {
            HttpHeaders headers = exchange.getRequest().getHeaders();
            MultiValueMap<String, String> queryParam = exchange.getRequest().getQueryParams();
            List<String> token = headers.get("token") != null ? headers.get("token") : queryParam.get("token");
​
            if (token == null || token.isEmpty()) return Mono.empty();
​
            return Mono.just(token.get(0) + "-" + exchange.getRequest().getURI().getPath());
        };
    }

这段代码就返回一个 KeyResolver 实现类对象，其中参数 exchange 是之前的文章中也频繁出现过的 ServerWebExchange，它是一个包含了请求所有信息的聚合头像，在这里你可以简单的将它理解为 HttpServletRequest。

我的代码内容比较简单，就是从请求中拿 token，我这里用 token 代替用户id来标识某个用户，如果能拿到就将 token 和请求路径拼接成字符串返回，如果不能拿到则返回空，返回值需要用Mono包装。

这样我就完成了一个策略 Key 对象的代码，然后就是将这个对象传入配置中就行了：.setKeyResolver(apiTokenKeyResolver()) 。

这个时候有的人可能就要问了：那你返回空怎么处理呢？如果你这个用户还没登录发出这个请求，这个时候你还拿不到token，那你是让它通过呢还是不通过呢？

这个问题我觉得非常好，就此就要引出了第三个配置：.setDenyEmptyKey(true)，这个配置参数是一个布尔值所以其入参只有两种可能，true 时代表返回空的请求禁止通过（403），false 则代表可以通过且不受限流限制。

4. 演示

演示之前我觉得有必要再回顾一下开头的配置：

    @Bean
    public RouteLocator routeLocator(RouteLocatorBuilder builder) {
        return builder.routes()
                .route((r) -> r.path("/user/**")
                        .filters(f -> f.requestRateLimiter()
                                .rateLimiter(RedisRateLimiter.class, c -> c.setReplenishRate(1).setBurstCapacity(10).setRequestedTokens(5))
                                .configure(c -> c.setKeyResolver(apiTokenKeyResolver()).setDenyEmptyKey(true)))
                        .uri("lb://user-api"))
                .build();
    }

这个时候再去看这个配置我相信你已经全然理解了这些配置所代表的具体含义是什么，但是我还是要简单的叙述一下它的意义：

声明一个每秒速率为1的令牌桶，每次请求消耗5个令牌，最大可容纳10个令牌，令牌桶所限制的资源是由请求中携带的token参数 + 请求路径拼接的字符串，当不携带token标识时，请求将被拒绝。

接着我直接通过浏览器去请求示例接口：

「微服务网关实战四」随意扩展定制的分布式限流，看看我怎么做

可以看到接口是通的，当我连续刷新浏览器时就会出现下面这种情况：

「微服务网关实战四」随意扩展定制的分布式限流，看看我怎么做

出现这种情况就代表我们的限流成功了，当一个请求被限流的时候会显示状态码 429，而且不返回任何数据，当然这个状态码是可以配置的，还记得一开始的配置截图吗？

「微服务网关实战四」随意扩展定制的分布式限流，看看我怎么做

可以在这里设置 StatusCode 这个字段来处理，而且你可以看到它的参数也是一个 HTTP 状态码的枚举。

除此之外我们再来看一个没有携带 token 的例子：

「微服务网关实战四」随意扩展定制的分布式限流，看看我怎么做

它直接返回了一个 403 同时也没有返回任何数据，这也是一个默认的处理，你也可以通过上图中的 emptyKeyStatus 字段去设置，虽然参数是一个字符串，但是不妨碍设置状态码，或者把设置 setDenyEmptyKey(false) 放行这种无资源标识的请求。

如果你想进一步了解其中的判断逻辑，可以参阅源码 RequestRateLimiterGatewayFilterFactory. apply 方法。

5. 最后

好了，今天的文章就是以上这些内容了，相关代码在这里，本篇主要的重点是网关限流的讲解，下一篇就要是第五讲了，将会献上本系列的重头戏：Nacos 动态配置网关路由。

通过我在网上博客的搜索，这块的内容讲解可以说是少之又少，相信一定不会让大家失望。

最后，如果大家觉得本文还不错的话就可以点赞以示支持，对内容有什么疑问也可以在评论区留言，我会积极对线的，下篇见。

本系列其他文章：

「微服务网关实战一」SCG 和 APISIX 该怎么选？

「微服务网关实战二」SCG + Nacos 动态感知上下线

「微服务网关实战三」详细理解 SCG 路由、断言与过滤器