导航：首页 > 开发技术 >

kubernetes怎么实现分布式限流

发表于：2024-11-26 作者：千家信息网编辑

千家信息网最后更新 2024年11月26日，本篇内容主要讲解"kubernetes怎么实现分布式限流"，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习"kubernetes怎么实现分布式限流"吧!一、概念

千家信息网最后更新 2024年11月26日kubernetes怎么实现分布式限流

本篇内容主要讲解"kubernetes怎么实现分布式限流"，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习"kubernetes怎么实现分布式限流"吧!

一、概念

限流(Ratelimiting)指对应用服务的请求进行限制，例如某一接口的请求限制为 100 个每秒,对超过限制的请求则进行快速失败或丢弃。

1.1 使用场景

限流可以应对：

热点业务带来的突发请求；
调用方 bug 导致的突发请求；
恶意攻击请求。

1.2 维度

对于限流场景，一般需要考虑两个维度的信息：
时间限流基于某段时间范围或者某个时间点，也就是我们常说的"时间窗口"，比如对每分钟、每秒钟的时间窗口做限定
资源基于可用资源的限制，比如设定最大访问次数，或最高可用连接数。
限流就是在某个时间窗口对资源访问做限制，比如设定每秒最多100个访问请求。

1.3 分布式限流

分布式限流相比于单机限流，只是把限流频次分配到各个节点中，比如限制某个服务访问100qps，如果有10个节点，那么每个节点理论上能够平均被访问10次，如果超过了则进行频率限制。

二、分布式限流常用方案

基于Guava的客户端限流Guava是一个客户端组件，在其多线程模块下提供了以RateLimiter为首的几个限流支持类。它只能对"当前"服务进行限流，即它不属于分布式限流的解决方案。

网关层限流服务网关，作为整个分布式链路中的第一道关卡，承接了所有用户来访请求。我们在网关层进行限流，就可以达到了整体限流的目的了。目前，主流的网关层有以软件为代表的Nginx，还有Spring Cloud中的Gateway和Zuul这类网关层组件，也有以硬件为代表的F5。

中间件限流将限流信息存储在分布式环境中某个中间件里（比如Redis缓存），每个组件都可以从这里获取到当前时刻的流量统计，从而决定是拒绝服务还是放行流量。

限流组件目前也有一些开源组件提供了限流的功能，比如Sentinel就是一个不错的选择。Sentinel是阿里出品的开源组件，并且包含在了Spring Cloud Alibaba组件库中。Hystrix也具有限流的功能。

Guava的Ratelimiter设计实现相当不错，可惜只能支持单机，网关层限流如果是单机则不太满足高可用，并且分布式网关的话还是需要依赖中间件限流，而redis之类的网络通信需要占用一小部分的网络消耗。阿里的Sentinel也是同理，底层使用的是redis或者zookeeper，每次访问都需要调用一次redis或者zk的接口。那么在云原生场景下，我们有没有什么更好的办法呢？

对于极致追求高性能的服务不需要考虑熔断、降级来说，是需要尽量减少网络之间的IO，那么是否可以通过一个总限频然后分配到具体的单机里面去，在单机中实现平均的限流，比如限制某个ip的qps为100，服务总共有10个节点，那么平均到每个服务里就是10qps，此时就可以通过guava的ratelimiter来实现了，甚至说如果服务的节点动态调整，单个服务的qps也能动态调整。

三、基于kubernetes的分布式限流

在Spring Boot应用中，定义一个filter，获取请求参数里的key（ip、userId等），然后根据key来获取rateLimiter，其中，rateLimiter的创建由数据库定义的限频数和副本数来判断，最后，再通过rateLimiter.tryAcquire来判断是否可以通过。

3.1 kubernetes中的副本数

在实际的服务中，数据上报服务一般无法确定客户端的上报时间、上报量，特别是对于这种要求高性能，服务一般都会用到HPA来实现动态扩缩容，所以，需要去间隔一段时间去获取服务的副本数。

func CountDeploymentSize(namespace string, deploymentName string) *int32 {        deployment, err := client.AppsV1().Deployments(namespace).Get(context.TODO(), deploymentName, metav1.GetOptions{})        if err != nil {                return nil        }        return deployment.Spec.Replicas}

用法：GET host/namespaces/test/deployments/k8s-rest-api直接即可。

3.2 rateLimiter的创建

在RateLimiterService中定义一个LoadingCache，其中，key可以为ip、userId等，并且，在多线程的情况下，使用refreshAfterWrite只阻塞加载数据的线程，其他线程则返回旧数据，极致发挥缓存的作用。

private final LoadingCache loadingCache = Caffeine.newBuilder()        .maximumSize(10_000)        .refreshAfterWrite(20, TimeUnit.MINUTES)        .build(this::createRateLimit);//定义一个默认最小的QPSprivate static final Integer minQpsLimit = 3000;

之后是创建rateLimiter，获取总限频数totalLimit和副本数replicas，之后是自己所需的逻辑判断，可以根据totalLimit和replicas的情况来进行qps的限定。

public RateLimiter createRateLimit(String key) {    log.info("createRateLimit,key:{}", key);    int totalLimit = 获取总限频数，可以在数据库中定义    Integer replicas = kubernetesService.getDeploymentReplicas();    RateLimiter rateLimiter;    if (totalLimit > 0 && replicas == null) {        rateLimiter = RateLimiter.create(totalLimit);    } else if (totalLimit > 0) {        int nodeQpsLimit = totalLimit / replicas;        rateLimiter = RateLimiter.create(nodeQpsLimit > minQpsLimit ? nodeQpsLimit : minQpsLimit);    } else {        rateLimiter = RateLimiter.create(minQpsLimit);    }    log.info("create rateLimiter success,key:{},rateLimiter:{}", key, rateLimiter);    return rateLimiter;}

3.3 rateLimiter的获取

根据key获取RateLimiter，如果有特殊需求的话，需要判断key不存在的尝尽

public RateLimiter getRateLimiter(String key) {  return loadingCache.get(key);}

3.4 filter里的判断

最后一步，就是使用rateLimiter来进行限流，如果rateLimiter.tryAcquire()为true，则进行filterChain.doFilter(request, response)，如果为false，则返回HttpStatus.TOO_MANY_REQUESTS

public class RateLimiterFilter implements Filter {    @Resource    private RateLimiterService rateLimiterService;    @Override    public void doFilter(ServletRequest request, ServletResponse response, FilterChain filterChain) throws IOException, ServletException {        HttpServletRequest httpServletRequest = (HttpServletRequest) request;        HttpServletResponse httpServletResponse = (HttpServletResponse) response;        String key = httpServletRequest.getHeader("key");        RateLimiter rateLimiter = rateLimiterService.getRateLimiter(key);        if (rateLimiter != null) {            if (rateLimiter.tryAcquire()) {                filterChain.doFilter(request, response);            } else {                httpServletResponse.setStatus(HttpStatus.TOO_MANY_REQUESTS.value());            }        } else {            filterChain.doFilter(request, response);        }    }}