限流

限流是面試中的常見的面試題（尤其是大廠面試、高P面試）

注：本文以 PDF 持續(xù)更新，最新尼恩 架構筆記、面試題 的PDF文件，請到文末《技術自由圈》公號獲取

為什么要限流

簡單來說：

限流在很多場景中用來限制并發(fā)和請求量，比如說秒殺搶購，保護自身系統(tǒng)和下游系統(tǒng)不被巨型流量沖垮等。

以微博為例，例如某某明星公布了戀情，訪問從平時的50萬增加到了500萬，系統(tǒng)的規(guī)劃能力，最多可以支撐200萬訪問，那么就要執(zhí)行限流規(guī)則，保證是一個可用的狀態(tài)，不至于服務器崩潰，所有請求不可用。

參考圖譜

系統(tǒng)架構知識圖譜（一張價值10w的系統(tǒng)架構知識圖譜）

https://www.processon.com/view/link/60fb9421637689719d246739

秒殺系統(tǒng)的架構

https://www.processon.com/view/link/61148c2b1e08536191d8f92f

限流的思想

在保證可用的情況下盡可能多增加進入的人數，其余的人在排隊等待，或者返回友好提示，保證里面的進行系統(tǒng)的用戶可以正常使用，防止系統(tǒng)雪崩。

日常生活中，有哪些需要限流的地方?

像我旁邊有一個國家景區(qū)，平時可能根本沒什么人前往，但是一到五一或者春節(jié)就人滿為患，這時候景區(qū)管理人員就會實行一系列的政策來限制進入人流量，
為什么要限流呢?

假如景區(qū)能容納一萬人，現在進去了三萬人，勢必摩肩接踵，整不好還會有事故發(fā)生，這樣的結果就是所有人的體驗都不好，如果發(fā)生了事故景區(qū)可能還要關閉，導致對外不可用，這樣的后果就是所有人都覺得體驗糟糕透了。

限流的算法

限流算法很多，常見的有三類，分別是計數器算法、漏桶算法、令牌桶算法，下面逐一講解。

限流的手段通常有計數器、漏桶、令牌桶。注意限流和限速（所有請求都會處理）的差別，視
業(yè)務場景而定。

（1）計數器：

在一段時間間隔內（時間窗/時間區(qū)間），處理請求的最大數量固定，超過部分不做處理。

（2）漏桶：

漏桶大小固定，處理速度固定，但請求進入速度不固定（在突發(fā)情況請求過多時，會丟棄過多的請求）。

（3）令牌桶：

令牌桶的大小固定，令牌的產生速度固定，但是消耗令牌（即請求）速度不固定（可以應對一些某些時間請求過多的情況）；每個請求都會從令牌桶中取出令牌，如果沒有令牌則丟棄該次請求。

計數器算法

計數器限流定義：

在一段時間間隔內（時間窗/時間區(qū)間），處理請求的最大數量固定，超過部分不做處理。

簡單粗暴，比如指定線程池大小，指定數據庫連接池大小、nginx連接數等，這都屬于計數器算法。

計數器算法是限流算法里最簡單也是最容易實現的一種算法。

舉個例子，比如我們規(guī)定對于A接口，我們1分鐘的訪問次數不能超過100個。

那么我們可以這么做：

• 在一開 始的時候，我們可以設置一個計數器counter，每當一個請求過來的時候，counter就加1，如果counter的值大于100并且該請求與第一個請求的間隔時間還在1分鐘之內，那么說明請求數過多，拒絕訪問；
• 如果該請求與第一個請求的間隔時間大于1分鐘，且counter的值還在限流范圍內，那么就重置 counter，就是這么簡單粗暴。

計算器限流的實現

package com.crazymaker.springcloud.ratelimit;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

// 計速器 限速
@Slf4j
public class CounterLimiter
{

    // 起始時間
    private static long startTime = System.currentTimeMillis();
    // 時間區(qū)間的時間間隔 ms
    private static long interval = 1000;
    // 每秒限制數量
    private static long maxCount = 2;
    //累加器
    private static AtomicLong accumulator = new AtomicLong();

    // 計數判斷, 是否超出限制
    private static long tryAcquire(long taskId, int turn)
    {
        long nowTime = System.currentTimeMillis();
        //在時間區(qū)間之內
        if (nowTime < startTime + interval)
        {
            long count = accumulator.incrementAndGet();

            if (count <= maxCount)
            {
                return count;
            } else
            {
                return -count;
            }
        } else
        {
            //在時間區(qū)間之外
            synchronized (CounterLimiter.class)
            {
                log.info("新時間區(qū)到了,taskId{}, turn {}..", taskId, turn);
                // 再一次判斷，防止重復初始化
                if (nowTime > startTime + interval)
                {
                    accumulator.set(0);
                    startTime = nowTime;
                }
            }
            return 0;
        }
    }

    //線程池，用于多線程模擬測試
    private ExecutorService pool = Executors.newFixedThreadPool(10);

    @Test
    public void testLimit()
    {

        // 被限制的次數
        AtomicInteger limited = new AtomicInteger(0);
        // 線程數
        final int threads = 2;
        // 每條線程的執(zhí)行輪數
        final int turns = 20;
        // 同步器
        CountDownLatch countDownLatch = new CountDownLatch(threads);
        long start = System.currentTimeMillis();
        for (int i = 0; i < threads; i++)
        {
            pool.submit(() ->
            {
                try
                {

                    for (int j = 0; j < turns; j++)
                    {

                        long taskId = Thread.currentThread().getId();
                        long index = tryAcquire(taskId, j);
                        if (index <= 0)
                        {
                            // 被限制的次數累積
                            limited.getAndIncrement();
                        }
                        Thread.sleep(200);
                    }


                } catch (Exception e)
                {
                    e.printStackTrace();
                }
                //等待所有線程結束
                countDownLatch.countDown();

            });
        }
        try
        {
            countDownLatch.await();
        } catch (InterruptedException e)
        {
            e.printStackTrace();
        }
        float time = (System.currentTimeMillis() - start) / 1000F;
        //輸出統(tǒng)計結果

        log.info("限制的次數為：" + limited.get() +
                ",通過的次數為：" + (threads * turns - limited.get()));
        log.info("限制的比例為：" + (float) limited.get() / (float) (threads * turns));
        log.info("運行的時長為：" + time);
    }


}

計數器限流的嚴重問題

這個算法雖然簡單，但是有一個十分致命的問題，那就是臨界問題，我們看下圖：

從上圖中我們可以看到，假設有一個惡意用戶，他在0:59時，瞬間發(fā)送了100個請求，并且1:00又瞬間發(fā)送了100個請求，那么其實這個用戶在 1秒里面，瞬間發(fā)送了200個請求。

我們剛才規(guī)定的是1分鐘最多100個請求（規(guī)劃的吞吐量），也就是每秒鐘最多1.7個請求，用戶通過在時間窗口的重置節(jié)點處突發(fā)請求， 可以瞬間超過我們的速率限制。

用戶有可能通過算法的這個漏洞，瞬間壓垮我們的應用。

說明：本文會持續(xù)更新，更多最新尼恩3高筆記PDF，請從下面的鏈接獲?。?碼云

漏桶算法

漏桶算法限流的基本原理為：水（對應請求）從進水口進入到漏桶里，漏桶以一定的速度出水（請求放行），當水流入速度過大，桶內的總水量大于桶容量會直接溢出，請求被拒絕，如圖所示。
大致的漏桶限流規(guī)則如下：
（1）進水口（對應客戶端請求）以任意速率流入進入漏桶。
（2）漏桶的容量是固定的，出水（放行）速率也是固定的。
（3）漏桶容量是不變的，如果處理速度太慢，桶內水量會超出了桶的容量，則后面流入的水滴會溢出，表示請求拒絕。

漏桶算法原理

漏桶算法思路很簡單：

水（請求）先進入到漏桶里，漏桶以一定的速度出水，當水流入速度過大會超過桶可接納的容量時直接溢出。

可以看出漏桶算法能強行限制數據的傳輸速率。

漏桶算法其實很簡單，可以粗略的認為就是注水漏水過程，往桶中以任意速率流入水，以一定速率流出水，當水超過桶容量（capacity）則丟棄，因為桶容量是不變的，保證了整體的速率。

以一定速率流出水，

削峰：有大量流量進入時，會發(fā)生溢出，從而限流保護服務可用

緩沖：不至于直接請求到服務器，緩沖壓力

消費速度固定 因為計算性能固定

漏桶算法實現

package com.crazymaker.springcloud.ratelimit;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

// 漏桶 限流
@Slf4j
public class LeakBucketLimiter {

    // 計算的起始時間
    private static long lastOutTime = System.currentTimeMillis();
    // 流出速率 每秒 2 次
    private static int leakRate = 2;

    // 桶的容量
    private static int capacity = 2;

    //剩余的水量
    private static AtomicInteger water = new AtomicInteger(0);

    //返回值說明：
    // false 沒有被限制到
    // true 被限流
    public static synchronized boolean isLimit(long taskId, int turn) {
        // 如果是空桶，就當前時間作為漏出的時間
        if (water.get() == 0) {
            lastOutTime = System.currentTimeMillis();
            water.addAndGet(1);
            return false;
        }
        // 執(zhí)行漏水
        int waterLeaked = ((int) ((System.currentTimeMillis() - lastOutTime) / 1000)) * leakRate;
        // 計算剩余水量
        int waterLeft = water.get() - waterLeaked;
        water.set(Math.max(0, waterLeft));
        // 重新更新leakTimeStamp
        lastOutTime = System.currentTimeMillis();
        // 嘗試加水,并且水還未滿 ，放行
        if ((water.get()) < capacity) {
            water.addAndGet(1);
            return false;
        } else {
            // 水滿，拒絕加水， 限流
            return true;
        }

    }


    //線程池，用于多線程模擬測試
    private ExecutorService pool = Executors.newFixedThreadPool(10);

    @Test
    public void testLimit() {

        // 被限制的次數
        AtomicInteger limited = new AtomicInteger(0);
        // 線程數
        final int threads = 2;
        // 每條線程的執(zhí)行輪數
        final int turns = 20;
        // 線程同步器
        CountDownLatch countDownLatch = new CountDownLatch(threads);
        long start = System.currentTimeMillis();
        for (int i = 0; i < threads; i++) {
            pool.submit(() ->
            {
                try {

                    for (int j = 0; j < turns; j++) {

                        long taskId = Thread.currentThread().getId();
                        boolean intercepted = isLimit(taskId, j);
                        if (intercepted) {
                            // 被限制的次數累積
                            limited.getAndIncrement();
                        }
                        Thread.sleep(200);
                    }


                } catch (Exception e) {
                    e.printStackTrace();
                }
                //等待所有線程結束
                countDownLatch.countDown();

            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        float time = (System.currentTimeMillis() - start) / 1000F;
        //輸出統(tǒng)計結果

        log.info("限制的次數為：" + limited.get() +
                ",通過的次數為：" + (threads * turns - limited.get()));
        log.info("限制的比例為：" + (float) limited.get() / (float) (threads * turns));
        log.info("運行的時長為：" + time);
    }
}

漏桶的問題

漏桶的出水速度固定，也就是請求放行速度是固定的。

網上抄來抄去的說法：

漏桶不能有效應對突發(fā)流量，但是能起到平滑突發(fā)流量（整流）的作用。

實際上的問題：

漏桶出口的速度固定，不能靈活的應對后端能力提升。比如，通過動態(tài)擴容，后端流量從1000QPS提升到1WQPS，漏桶沒有辦法。

令牌桶限流

令牌桶算法以一個設定的速率產生令牌并放入令牌桶，每次用戶請求都得申請令牌，如果令牌不足，則拒絕請求。
令牌桶算法中新請求到來時會從桶里拿走一個令牌，如果桶內沒有令牌可拿，就拒絕服務。當然，令牌的數量也是有上限的。令牌的數量與時間和發(fā)放速率強相關，時間流逝的時間越長，會不斷往桶里加入越多的令牌，如果令牌發(fā)放的速度比申請速度快，令牌桶會放滿令牌，直到令牌占滿整個令牌桶，如圖所示。

令牌桶限流大致的規(guī)則如下：
（1）進水口按照某個速度，向桶中放入令牌。
（2）令牌的容量是固定的，但是放行的速度不是固定的，只要桶中還有剩余令牌，一旦請求過來就能申請成功，然后放行。
（3）如果令牌的發(fā)放速度，慢于請求到來速度，桶內就無牌可領，請求就會被拒絕。

總之，令牌的發(fā)送速率可以設置，從而可以對突發(fā)的出口流量進行有效的應對。

令牌桶算法

令牌桶與漏桶相似，不同的是令牌桶桶中放了一些令牌，服務請求到達后，要獲取令牌之后才會得到服務，舉個例子，我們平時去食堂吃飯，都是在食堂內窗口前排隊的，這就好比是漏桶算法，大量的人員聚集在食堂內窗口外，以一定的速度享受服務，如果涌進來的人太多，食堂裝不下了，可能就有一部分人站到食堂外了，這就沒有享受到食堂的服務，稱之為溢出，溢出可以繼續(xù)請求，也就是繼續(xù)排隊，那么這樣有什么問題呢?

如果這時候有特殊情況，比如有些趕時間的志愿者啦、或者高三要高考啦，這種情況就是突發(fā)情況，如果也用漏桶算法那也得慢慢排隊，這也就沒有解決我們的需求，對于很多應用場景來說，除了要求能夠限制數據的平均傳輸速率外，還要求允許某種程度的突發(fā)傳輸。這時候漏桶算法可能就不合適了，令牌桶算法更為適合。如圖所示，令牌桶算法的原理是系統(tǒng)會以一個恒定的速度往桶里放入令牌，而如果請求需要被處理，則需要先從桶里獲取一個令牌，當桶里沒有令牌可取時，則拒絕服務。

令牌桶算法實現

package com.crazymaker.springcloud.ratelimit;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

// 令牌桶 限速
@Slf4j
public class TokenBucketLimiter {
    // 上一次令牌發(fā)放時間
    public long lastTime = System.currentTimeMillis();
    // 桶的容量
    public int capacity = 2;
    // 令牌生成速度 /s
    public int rate = 2;
    // 當前令牌數量
    public AtomicInteger tokens = new AtomicInteger(0);
    ;

    //返回值說明：
    // false 沒有被限制到
    // true 被限流
    public synchronized boolean isLimited(long taskId, int applyCount) {
        long now = System.currentTimeMillis();
        //時間間隔,單位為 ms
        long gap = now - lastTime;

        //計算時間段內的令牌數
        int reverse_permits = (int) (gap * rate / 1000);
        int all_permits = tokens.get() + reverse_permits;
        // 當前令牌數
        tokens.set(Math.min(capacity, all_permits));
        log.info("tokens {} capacity {} gap {} ", tokens, capacity, gap);

        if (tokens.get() < applyCount) {
            // 若拿不到令牌,則拒絕
            // log.info("被限流了.." + taskId + ", applyCount: " + applyCount);
            return true;
        } else {
            // 還有令牌，領取令牌
            tokens.getAndAdd( - applyCount);
            lastTime = now;

            // log.info("剩余令牌.." + tokens);
            return false;
        }

    }

    //線程池，用于多線程模擬測試
    private ExecutorService pool = Executors.newFixedThreadPool(10);

    @Test
    public void testLimit() {

        // 被限制的次數
        AtomicInteger limited = new AtomicInteger(0);
        // 線程數
        final int threads = 2;
        // 每條線程的執(zhí)行輪數
        final int turns = 20;


        // 同步器
        CountDownLatch countDownLatch = new CountDownLatch(threads);
        long start = System.currentTimeMillis();
        for (int i = 0; i < threads; i++) {
            pool.submit(() ->
            {
                try {

                    for (int j = 0; j < turns; j++) {

                        long taskId = Thread.currentThread().getId();
                        boolean intercepted = isLimited(taskId, 1);
                        if (intercepted) {
                            // 被限制的次數累積
                            limited.getAndIncrement();
                        }

                        Thread.sleep(200);
                    }


                } catch (Exception e) {
                    e.printStackTrace();
                }
                //等待所有線程結束
                countDownLatch.countDown();

            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        float time = (System.currentTimeMillis() - start) / 1000F;
        //輸出統(tǒng)計結果

        log.info("限制的次數為：" + limited.get() +
                ",通過的次數為：" + (threads * turns - limited.get()));
        log.info("限制的比例為：" + (float) limited.get() / (float) (threads * turns));
        log.info("運行的時長為：" + time);
    }
}

令牌桶的好處

令牌桶的好處之一就是可以方便地應對 突發(fā)出口流量（后端能力的提升）。

比如，可以改變令牌的發(fā)放速度，算法能按照新的發(fā)送速率調大令牌的發(fā)放數量，使得出口突發(fā)流量能被處理。

Guava RateLimiter

Guava是Java領域優(yōu)秀的開源項目，它包含了Google在Java項目中使用一些核心庫，包含集合(Collections)，緩存(Caching)，并發(fā)編程庫(Concurrency)，常用注解(Common annotations)，String操作，I/O操作方面的眾多非常實用的函數。 Guava的 RateLimiter提供了令牌桶算法實現：平滑突發(fā)限流(SmoothBursty)和平滑預熱限流(SmoothWarmingUp)實現。

RateLimiter的類圖如上所示，

Nginx漏桶限流

Nginx限流的簡單演示

每六秒才處理一次請求，如下

  limit_req_zone  $arg_sku_id  zone=skuzone:10m      rate=6r/m;
  limit_req_zone  $http_user_id  zone=userzone:10m      rate=6r/m;
  limit_req_zone  $binary_remote_addr  zone=perip:10m      rate=6r/m;
  limit_req_zone  $server_name        zone=perserver:1m   rate=6r/m;

這是從請求參數里邊，提前參數做限流

這是從請求參數里邊，提前參數，進行限流的次數統(tǒng)計key。

在http塊里邊定義限流的內存區(qū)域 zone。

  limit_req_zone  $arg_sku_id  zone=skuzone:10m      rate=6r/m;
  limit_req_zone  $http_user_id  zone=userzone:10m      rate=6r/m;
  limit_req_zone  $binary_remote_addr  zone=perip:10m      rate=6r/m;
  limit_req_zone  $server_name        zone=perserver:1m   rate=10r/s;

在location塊中使用 限流zone，參考如下：

#  ratelimit by sku id
    location  = /ratelimit/sku {
      limit_req  zone=skuzone;
      echo "正常的響應";
}

測試

[root@cdh1 ~]# /vagrant/LuaDemoProject/sh/linux/openresty-restart.sh
shell dir is: /vagrant/LuaDemoProject/sh/linux
Shutting down openrestry/nginx:  pid is 13479 13485
Shutting down  succeeded!
OPENRESTRY_PATH:/usr/local/openresty
PROJECT_PATH:/vagrant/LuaDemoProject/src
nginx: [alert] lua_code_cache is off; this will hurt performance in /vagrant/LuaDemoProject/src/conf/nginx-seckill.conf:90
openrestry/nginx starting succeeded!
pid is 14197


[root@cdh1 ~]# curl  http://cdh1/ratelimit/sku?sku_id=1
正常的響應
root@cdh1 ~]#  curl  http://cdh1/ratelimit/sku?sku_id=1
正常的響應
[root@cdh1 ~]#  curl  http://cdh1/ratelimit/sku?sku_id=1
限流后的降級內容
[root@cdh1 ~]#  curl  http://cdh1/ratelimit/sku?sku_id=1
限流后的降級內容
[root@cdh1 ~]#  curl  http://cdh1/ratelimit/sku?sku_id=1
限流后的降級內容
[root@cdh1 ~]#  curl  http://cdh1/ratelimit/sku?sku_id=1
限流后的降級內容
[root@cdh1 ~]#  curl  http://cdh1/ratelimit/sku?sku_id=1
限流后的降級內容
[root@cdh1 ~]#  curl  http://cdh1/ratelimit/sku?sku_id=1
正常的響應

從Header頭部提前參數

1、nginx是支持讀取非nginx標準的用戶自定義header的，但是需要在http或者server下開啟header的下劃線支持:

underscores_in_headers on;

2、比如我們自定義header為X-Real-IP，通過第二個nginx獲取該header時需要這樣:

$http_x_real_ip; (一律采用小寫，而且前面多了個http_)

 underscores_in_headers on;

  limit_req_zone  $http_user_id  zone=userzone:10m      rate=6r/m;
  server {
    listen       80 default;
    server_name  nginx.server *.nginx.server;
    default_type 'text/html';
    charset utf-8;


#  ratelimit by user id
    location  = /ratelimit/demo {
      limit_req  zone=userzone;
      echo "正常的響應";
    }


  
    location = /50x.html{
      echo "限流后的降級內容";
    }

    error_page 502 503 =200 /50x.html;

  }

測試

[root@cdh1 ~]# curl -H "USER-ID:1" http://cdh1/ratelimit/demo
正常的響應
[root@cdh1 ~]# curl -H "USER-ID:1" http://cdh1/ratelimit/demo
限流后的降級內容
[root@cdh1 ~]# curl -H "USER-ID:1" http://cdh1/ratelimit/demo
限流后的降級內容
[root@cdh1 ~]# curl -H "USER-ID:1" http://cdh1/ratelimit/demo
限流后的降級內容
[root@cdh1 ~]# curl -H "USER-ID:1" http://cdh1/ratelimit/demo
限流后的降級內容
[root@cdh1 ~]# curl -H "USER-ID:1" http://cdh1/ratelimit/demo
限流后的降級內容
[root@cdh1 ~]# curl -H "USER-ID:1" http://cdh1/ratelimit/demo
限流后的降級內容
[root@cdh1 ~]# curl -H "USER_ID:2" http://cdh1/ratelimit/demo
正常的響應
[root@cdh1 ~]# curl -H "USER_ID:2" http://cdh1/ratelimit/demo
限流后的降級內容
[root@cdh1 ~]#
[root@cdh1 ~]# curl -H "USER_ID:2" http://cdh1/ratelimit/demo
限流后的降級內容
[root@cdh1 ~]# curl -H "USER-ID:3" http://cdh1/ratelimit/demo
正常的響應
[root@cdh1 ~]# curl -H "USER-ID:3" http://cdh1/ratelimit/demo
限流后的降級內容

Nginx漏桶限流的三個細分類型，即burst、nodelay參數詳解

每六秒才處理一次請求，如下

limit_req_zone  $arg_user_id  zone=limti_req_zone:10m      rate=10r/m;

不帶緩沖隊列的漏桶限流

limit_req zone=limti_req_zone;

• 嚴格依照在limti_req_zone中配置的rate來處理請求
• 超過rate處理能力范圍的，直接drop
• 表現為對收到的請求無延時

假設1秒內提交10個請求，可以看到一共10個請求，9個請求都失敗了，直接返回503，

接著再查看 /var/log/nginx/access.log，印證了只有一個請求成功了，其它就是都直接返回了503，即服務器拒絕了請求。

帶緩沖隊列的漏桶限流

limit_req zone=limti_req_zone burst=5;

• 依照在limti_req_zone中配置的rate來處理請求
• 同時設置了一個大小為5的緩沖隊列，在緩沖隊列中的請求會等待慢慢處理
• 超過了burst緩沖隊列長度和rate處理能力的請求被直接丟棄
• 表現為對收到的請求有延時

假設1秒內提交10個請求，則可以發(fā)現在1s內，**在服務器接收到10個并發(fā)請求后，先處理1個請求，同時將5個請求放入burst緩沖隊列中，等待處理。而超過（burst+1）數量的請求就被直接拋棄了，即直接拋棄了4個請求。**burst緩存的5個請求每隔6s處理一次。

接著查看 /var/log/nginx/access.log日志

帶瞬時處理能力的漏桶限流

limit_req zone=req_zone burst=5 nodelay;

如果設置nodelay，會在瞬時提供處理(burst + rate)個請求的能力，請求數量超過**（burst + rate）**的時候就會直接返回503，峰值范圍內的請求，不存在請求需要等待的情況。

假設1秒內提交10個請求，則可以發(fā)現在1s內，服務器端處理了6個請求（峰值速度：burst＋10s內一個請求）。對于剩下的4個請求，直接返回503，在下一秒如果繼續(xù)向服務端發(fā)送10個請求，服務端會直接拒絕這10個請求并返回503。

接著查看 /var/log/nginx/access.log日志

可以發(fā)現在1s內，服務器端處理了6個請求（峰值速度：burst＋原來的處理速度）。對于剩下的4個請求，直接返回503。

但是，總數額度和速度*時間保持一致， 就是額度用完了，需要等到一個有額度的時間段，才開始接收新的請求。如果一次處理了5個請求，相當于占了30s的額度，6*5=30。因為設定了6s處理1個請求，所以直到30
s 之后，才可以再處理一個請求，即如果此時向服務端發(fā)送10個請求，會返回9個503，一個200

說明：本文會以pdf格式持續(xù)更新，更多最新尼恩3高pdf筆記，請從下面的鏈接獲取：語雀 或者 碼云

分布式限流組件

why

但是Nginx的限流指令只能在同一塊內存區(qū)域有效，而在生產場景中秒殺的外部網關往往是多節(jié)點部署，所以這就需要用到分布式限流組件。

高性能的分布式限流組件可以使用Redis+Lua來開發(fā)，京東的搶購就是使用Redis+Lua完成的限流。并且無論是Nginx外部網關還是Zuul內部網關，都可以使用Redis+Lua限流組件。

理論上，接入層的限流有多個維度：

（1）用戶維度限流：在某一時間段內只允許用戶提交一次請求，比如可以采取客戶端IP或者用戶ID作為限流的key。

（2）商品維度的限流：對于同一個搶購商品，在某個時間段內只允許一定數量的請求進入，可以采取秒殺商品ID作為限流的key。

什么時候用nginx限流：

用戶維度的限流，可以在ngix 上進行，因為使用nginx限流內存來存儲用戶id，比用redis 的key，來存儲用戶id，效率高。

什么時候用redis+lua分布式限流：

商品維度的限流，可以在redis上進行，不需要大量的計算訪問次數的key，另外，可以控制所有的接入層節(jié)點的訪問秒殺請求的總量。

redis+lua分布式限流組件

--- 此腳本的環(huán)境： redis 內部，不是運行在 nginx 內部

---方法：申請令牌
--- -1 failed
--- 1 success
--- @param key key 限流關鍵字
--- @param apply  申請的令牌數量
local function acquire(key, apply)
    local times = redis.call('TIME');
    -- times[1] 秒數   -- times[2] 微秒數
    local curr_mill_second = times[1] * 1000000 + times[2];
    curr_mill_second = curr_mill_second / 1000;

    local cacheInfo = redis.pcall("HMGET", key, "last_mill_second", "curr_permits", "max_permits", "rate")
    --- 局部變量：上次申請的時間
    local last_mill_second = cacheInfo[1];
    --- 局部變量：之前的令牌數
    local curr_permits = tonumber(cacheInfo[2]);
    --- 局部變量：桶的容量
    local max_permits = tonumber(cacheInfo[3]);
    --- 局部變量：令牌的發(fā)放速率
    local rate = cacheInfo[4];
    --- 局部變量：本次的令牌數
    local local_curr_permits = 0;

    if (type(last_mill_second) ~= 'boolean' and last_mill_second ~= nil) then
        -- 計算時間段內的令牌數
        local reverse_permits = math.floor(((curr_mill_second - last_mill_second) / 1000) * rate);
        -- 令牌總數
        local expect_curr_permits = reverse_permits + curr_permits;
        -- 可以申請的令牌總數
        local_curr_permits = math.min(expect_curr_permits, max_permits);
    else
        -- 第一次獲取令牌
        redis.pcall("HSET", key, "last_mill_second", curr_mill_second)
        local_curr_permits = max_permits;
    end

    local result = -1;
    -- 有足夠的令牌可以申請
    if (local_curr_permits - apply >= 0) then
        -- 保存剩余的令牌
        redis.pcall("HSET", key, "curr_permits", local_curr_permits - apply);
        -- 為下次的令牌獲取，保存時間
        redis.pcall("HSET", key, "last_mill_second", curr_mill_second)
        -- 返回令牌獲取成功
        result = 1;
    else
        -- 返回令牌獲取失敗
        result = -1;
    end
    return result
end
--eg
-- /usr/local/redis/bin/redis-cli  -a 123456  --eval   /vagrant/LuaDemoProject/src/luaScript/redis/rate_limiter.lua key , acquire 1  1

-- 獲取 sha編碼的命令
-- /usr/local/redis/bin/redis-cli  -a 123456  script load "$(cat  /vagrant/LuaDemoProject/src/luaScript/redis/rate_limiter.lua)"
-- /usr/local/redis/bin/redis-cli  -a 123456  script exists  "cf43613f172388c34a1130a760fc699a5ee6f2a9"

-- /usr/local/redis/bin/redis-cli -a 123456  evalsha   "cf43613f172388c34a1130a760fc699a5ee6f2a9" 1 "rate_limiter:seckill:1"  init 1  1
-- /usr/local/redis/bin/redis-cli -a 123456  evalsha   "cf43613f172388c34a1130a760fc699a5ee6f2a9" 1 "rate_limiter:seckill:1"  acquire 1

--local rateLimiterSha = "e4e49e4c7b23f0bf7a2bfee73e8a01629e33324b";

---方法：初始化限流 Key
--- 1 success
--- @param key key
--- @param max_permits  桶的容量
--- @param rate  令牌的發(fā)放速率
local function init(key, max_permits, rate)
    local rate_limit_info = redis.pcall("HMGET", key, "last_mill_second", "curr_permits", "max_permits", "rate")
    local org_max_permits = tonumber(rate_limit_info[3])
    local org_rate = rate_limit_info[4]

    if (org_max_permits == nil) or (rate ~= org_rate or max_permits ~= org_max_permits) then
        redis.pcall("HMSET", key, "max_permits", max_permits, "rate", rate, "curr_permits", max_permits)
    end
    return 1;
end
--eg
-- /usr/local/redis/bin/redis-cli -a 123456 --eval   /vagrant/LuaDemoProject/src/luaScript/redis/rate_limiter.lua key , init 1  1
-- /usr/local/redis/bin/redis-cli -a 123456 --eval   /vagrant/LuaDemoProject/src/luaScript/redis/rate_limiter.lua  "rate_limiter:seckill:1"  , init 1  1


---方法：刪除限流 Key
local function delete(key)
    redis.pcall("DEL", key)
    return 1;
end
--eg
-- /usr/local/redis/bin/redis-cli  --eval   /vagrant/LuaDemoProject/src/luaScript/redis/rate_limiter.lua key , delete


local key = KEYS[1]
local method = ARGV[1]
if method == 'acquire' then
    return acquire(key, ARGV[2], ARGV[3])
elseif method == 'init' then
    return init(key, ARGV[2], ARGV[3])
elseif method == 'delete' then
    return delete(key)
else
    --ignore
end

在redis中，為了避免重復發(fā)送腳本數據浪費網絡資源，可以使用script load命令進行腳本數據緩存，并且返回一個哈希碼作為腳本的調用句柄，

每次調用腳本只需要發(fā)送哈希碼來調用即可。

分布式令牌限流實戰(zhàn)

可以使用redis+lua，實戰(zhàn)一票下邊的簡單案例：

令牌按照1個每秒的速率放入令牌桶，桶中最多存放2個令牌，那系統(tǒng)就只會允許持續(xù)的每秒處理2個請求，

或者每隔2 秒，等桶中2 個令牌攢滿后，一次處理2個請求的突發(fā)情況，保證系統(tǒng)穩(wěn)定性。

商品維度的限流

當秒殺商品維度的限流，當商品的流量，遠遠大于涉及的流量時，開始隨機丟棄請求。

Nginx的令牌桶限流腳本getToken_access_limit.lua執(zhí)行在請求的access階段，但是，該腳本并沒有實現限流的核心邏輯，僅僅調用緩存在Redis內部的rate_limiter.lua腳本進行限流。

getToken_access_limit.lua腳本和rate_limiter.lua腳本的關系，具體如圖10-17所示。

圖10-17 getToken_access_limit.lua腳本和rate_limiter.lua腳本關系

什么時候在Redis中加載rate_limiter.lua腳本呢？

和秒殺腳本一樣，該腳本是在Java程序啟動商品秒殺時，完成其在Redis的加載和緩存的。

還有一點非常重要，Java程序會將腳本加載完成之后的sha1編碼，去通過自定義的key（具體為"lua:sha1:rate_limiter"）緩存在Redis中，以方便Nginx的getToken_access_limit.lua腳本去獲取，并且在調用evalsha方法時使用。

注意：使用redis集群，因此每個節(jié)點都需要各自緩存一份腳本數據

/**
* 由于使用redis集群，因此每個節(jié)點都需要各自緩存一份腳本數據
* @param slotKey 用來定位對應的slot的slotKey
*/
public void storeScript(String slotKey){
if (StringUtils.isEmpty(unlockSha1) || !jedisCluster.scriptExists(unlockSha1, slotKey)){
   //redis支持腳本緩存，返回哈希碼，后續(xù)可以繼續(xù)用來調用腳本
    unlockSha1 = jedisCluster.scriptLoad(DISTRIBUTE_LOCK_SCRIPT_UNLOCK_VAL, slotKey);
   }
}

常見的限流組件

redission分布式限流采用令牌桶思想和固定時間窗口，trySetRate方法設置桶的大小，利用redis key過期機制達到時間窗口目的，控制固定時間窗口內允許通過的請求量。

spring cloud gateway集成redis限流，但屬于網關層限流

技術自由的實現路徑：

實現你的 架構自由：

《吃透8圖1模板，人人可以做架構》

《10Wqps評論中臺，如何架構？B站是這么做的?。?！》

《阿里二面：千萬級、億級數據，如何性能優(yōu)化？ 教科書級 答案來了》

《峰值21WQps、億級DAU，小游戲《羊了個羊》是怎么架構的？》

《100億級訂單怎么調度，來一個大廠的極品方案》

《2個大廠 100億級 超大流量 紅包 架構方案》

… 更多架構文章，正在添加中

實現你的 響應式 自由：

《響應式圣經：10W字，實現Spring響應式編程自由》

這是老版本 《Flux、Mono、Reactor 實戰(zhàn)（史上最全）》

實現你的 spring cloud 自由：

《Spring cloud Alibaba 學習圣經》 PDF

《分庫分表 Sharding-JDBC 底層原理、核心實戰(zhàn)（史上最全）》

《一文搞定：SpringBoot、SLF4j、Log4j、Logback、Netty之間混亂關系（史上最全）》

實現你的 linux 自由：

《Linux命令大全：2W多字，一次實現Linux自由》

實現你的 網絡 自由：

《TCP協議詳解 (史上最全)》

《網絡三張表：ARP表, MAC表, 路由表，實現你的網絡自由?。　?/p>

實現你的 分布式鎖 自由：

《Redis分布式鎖（圖解 - 秒懂 - 史上最全）》

《Zookeeper 分布式鎖 - 圖解 - 秒懂》

實現你的 王者組件 自由：

《隊列之王： Disruptor 原理、架構、源碼 一文穿透》

《緩存之王：Caffeine 源碼、架構、原理（史上最全，10W字 超級長文）》

《緩存之王：Caffeine 的使用（史上最全）》

《Java Agent 探針、字節(jié)碼增強 ByteBuddy（史上最全）》

實現你的 面試題 自由：

4000頁《尼恩Java面試寶典 》 40個專題

以上尼恩 架構筆記、面試題 的PDF文件更新，請到《技術自由圈》公號獲取↓↓↓文章來源地址http://www.zghlxwxcb.cn/news/detail-415878.html

到了這里，關于限流：計數器、漏桶、令牌桶 三大算法的原理與實戰(zhàn)（史上最全）的文章就介紹完了。如果您還想了解更多內容，請在右上角搜索TOY模板網以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網！