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<String, RateLimiter>，其中，key可以為ip、userId等，并且，在多線程的情況下，使用refreshAfterWrite只阻塞加載數據的線程，其他線程則返回舊數據，極致發揮緩存的作用。

private final LoadingCache<String, RateLimiter> loadingCache = Caffeine.newBuilder()
        .maximumSize(10_000)
        .refreshAfterWrite(20, TimeUnit.MINUTES)
        .build(this::createRateLimit);
//定義一個默認最小的QPS
private 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);
        }
    }
}

四、性能壓測

為了方便對比性能之間的差距，我們在本地單機做了下列測試，其中，總限頻都設置為3萬。

無限流

使用redis限流

其中，ping redis大概6-7ms左右，對應的，每次請求需要訪問redis，時延都有大概6-7ms，性能下降明顯

自研限流

性能幾乎追平無限流的場景，guava的rateLimiter確實表現卓越

五、其他問題

5.1 對于保證qps限頻準確的時候，應該怎么解決呢？

在k8s中，服務是動態擴縮容的，相應的，每個節點應該都要有所變化，如果對外宣稱限頻100qps，而且后續業務方真的要求百分百準確，只能把LoadingCache<String, RateLimiter>的過期時間調小一點，讓它能夠近實時的更新單節點的qps。這里還需要考慮一下k8s的壓力，因為每次都要獲取副本數，這里也是需要做緩存的

5.2 服務從1個節點動態擴為4個節點，這個時候新節點識別為4，但其實有些并沒有啟動完，會不會造成某個節點承受了太大的壓力

理論上是存在這個可能的，這個時候需要考慮一下初始的副本數的，擴縮容不能一蹴而就，一下子從1變為4變為幾十個這種。一般的話，生產環境肯定是不能只有一個節點，并且要考慮擴縮容的話，至于要有多個副本預備的

5.3 如果有多個副本，怎么保證請求是均勻的

這個是依賴于k8s的service負載均衡策略的，這個我們之前做過實驗，流量確實是能夠均勻的落到節點上的。還有就是，我們整個限流都是基于k8s的，如果k8s出現問題，那就是整個集群所有服務都有可能出現問題了。

到此，相信大家對“kubernetes怎么實現分布式限流”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！