一、概念

限流(Ratelimiting)指對應(yīng)用服務(wù)的請求進(jìn)行限制，例如某一接口的請求限制為 100 個每秒,對超過限制的請求則進(jìn)行快速失敗或丟棄。

1.1 使用場景

限流可以應(yīng)對：

• 熱點業(yè)務(wù)帶來的突發(fā)請求；
• 調(diào)用方 bug 導(dǎo)致的突發(fā)請求；
• 惡意攻擊請求。

1.2 維度

對于限流場景，一般需要考慮兩個維度的信息：
時間限流基于某段時間范圍或者某個時間點，也就是我們常說的“時間窗口”，比如對每分鐘、每秒鐘的時間窗口做限定
資源基于可用資源的限制，比如設(shè)定最大訪問次數(shù)，或最高可用連接數(shù)。
限流就是在某個時間窗口對資源訪問做限制，比如設(shè)定每秒最多100個訪問請求。

1.3 分布式限流

分布式限流相比于單機(jī)限流，只是把限流頻次分配到各個節(jié)點中，比如限制某個服務(wù)訪問100qps，如果有10個節(jié)點，那么每個節(jié)點理論上能夠平均被訪問10次，如果超過了則進(jìn)行頻率限制。

二、分布式限流常用方案

基于Guava的客戶端限流Guava是一個客戶端組件，在其多線程模塊下提供了以RateLimiter為首的幾個限流支持類。它只能對“當(dāng)前”服務(wù)進(jìn)行限流，即它不屬于分布式限流的解決方案。

網(wǎng)關(guān)層限流服務(wù)網(wǎng)關(guān)，作為整個分布式鏈路中的第一道關(guān)卡，承接了所有用戶來訪請求。我們在網(wǎng)關(guān)層進(jìn)行限流，就可以達(dá)到了整體限流的目的了。目前，主流的網(wǎng)關(guān)層有以軟件為代表的Nginx，還有Spring Cloud中的Gateway和Zuul這類網(wǎng)關(guān)層組件，也有以硬件為代表的F5。

中間件限流將限流信息存儲在分布式環(huán)境中某個中間件里（比如Redis緩存），每個組件都可以從這里獲取到當(dāng)前時刻的流量統(tǒng)計，從而決定是拒絕服務(wù)還是放行流量。

限流組件目前也有一些開源組件提供了限流的功能，比如Sentinel就是一個不錯的選擇。Sentinel是阿里出品的開源組件，并且包含在了Spring Cloud Alibaba組件庫中。Hystrix也具有限流的功能。

Guava的Ratelimiter設(shè)計實現(xiàn)相當(dāng)不錯，可惜只能支持單機(jī)，網(wǎng)關(guān)層限流如果是單機(jī)則不太滿足高可用，并且分布式網(wǎng)關(guān)的話還是需要依賴中間件限流，而redis之類的網(wǎng)絡(luò)通信需要占用一小部分的網(wǎng)絡(luò)消耗。阿里的Sentinel也是同理，底層使用的是redis或者zookeeper，每次訪問都需要調(diào)用一次redis或者zk的接口。那么在云原生場景下，我們有沒有什么更好的辦法呢？

對于極致追求高性能的服務(wù)不需要考慮熔斷、降級來說，是需要盡量減少網(wǎng)絡(luò)之間的IO，那么是否可以通過一個總限頻然后分配到具體的單機(jī)里面去，在單機(jī)中實現(xiàn)平均的限流，比如限制某個ip的qps為100，服務(wù)總共有10個節(jié)點，那么平均到每個服務(wù)里就是10qps，此時就可以通過guava的ratelimiter來實現(xiàn)了，甚至說如果服務(wù)的節(jié)點動態(tài)調(diào)整，單個服務(wù)的qps也能動態(tài)調(diào)整。

三、基于kubernetes的分布式限流

在Spring Boot應(yīng)用中，定義一個filter，獲取請求參數(shù)里的key（ip、userId等），然后根據(jù)key來獲取rateLimiter，其中，rateLimiter的創(chuàng)建由數(shù)據(jù)庫定義的限頻數(shù)和副本數(shù)來判斷，最后，再通過rateLimiter.tryAcquire來判斷是否可以通過。

3.1 kubernetes中的副本數(shù)

在實際的服務(wù)中，數(shù)據(jù)上報服務(wù)一般無法確定客戶端的上報時間、上報量，特別是對于這種要求高性能，服務(wù)一般都會用到HPA來實現(xiàn)動態(tài)擴(kuò)縮容，所以，需要去間隔一段時間去獲取服務(wù)的副本數(shù)。

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的創(chuàng)建

在RateLimiterService中定義一個LoadingCache<String, RateLimiter>，其中，key可以為ip、userId等，并且，在多線程的情況下，使用refreshAfterWrite只阻塞加載數(shù)據(jù)的線程，其他線程則返回舊數(shù)據(jù)，極致發(fā)揮緩存的作用。

private final LoadingCache<String, RateLimiter> loadingCache = Caffeine.newBuilder()
        .maximumSize(10_000)
        .refreshAfterWrite(20, TimeUnit.MINUTES)
        .build(this::createRateLimit);
//定義一個默認(rèn)最小的QPS
private static final Integer minQpsLimit = 3000;

之后是創(chuàng)建rateLimiter，獲取總限頻數(shù)totalLimit和副本數(shù)replicas，之后是自己所需的邏輯判斷，可以根據(jù)totalLimit和replicas的情況來進(jìn)行qps的限定。

public RateLimiter createRateLimit(String key) {
    log.info("createRateLimit,key:{}", key);
    int totalLimit = 獲取總限頻數(shù)，可以在數(shù)據(jù)庫中定義
    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的獲取

根據(jù)key獲取RateLimiter，如果有特殊需求的話，需要判斷key不存在的嘗盡

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

3.4 filter里的判斷

最后一步，就是使用rateLimiter來進(jìn)行限流，如果rateLimiter.tryAcquire()為true，則進(jìn)行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);
        }
    }
}

四、性能壓測

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

無限流

使用redis限流

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

自研限流

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

五、其他問題

5.1 對于保證qps限頻準(zhǔn)確的時候，應(yīng)該怎么解決呢？

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

5.2 服務(wù)從1個節(jié)點動態(tài)擴(kuò)為4個節(jié)點，這個時候新節(jié)點識別為4，但其實有些并沒有啟動完，會不會造成某個節(jié)點承受了太大的壓力

理論上是存在這個可能的，這個時候需要考慮一下初始的副本數(shù)的，擴(kuò)縮容不能一蹴而就，一下子從1變?yōu)?變?yōu)閹资畟€這種。一般的話，生產(chǎn)環(huán)境肯定是不能只有一個節(jié)點，并且要考慮擴(kuò)縮容的話，至于要有多個副本預(yù)備的

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

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

到此這篇關(guān)于kubernetes實現(xiàn)分布式限流的文章就介紹到這了。希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

最新評論