摘要：算法簡介和示例說明業界比較流行的限流算法有漏桶算法和令牌桶算法。判斷接口是否限流其實就是看能不能從令牌桶中取出令牌，方法如下判斷接口是否被限流更新令牌桶狀態到了這里，相信讀者已經對令牌桶算法有了一個比較清晰的認識了。

我們開發的接口服務系統很多都具有抗高并發，保證高可用的特性。現實條件下，隨著流量的不斷增加，在經費、硬件和資源受限的情況下，我們就需要為我們的系統服務制定有效的限流、分流策略來保護我們的系統了。

業界比較流行的限流算法有漏桶算法和令牌桶算法。

漏桶(Leaky Bucket)算法的實現思路比較簡單，水(請求)先流入到桶中，然后桶以一定的速度出水(接口有響應速率)，當水流過大時(訪問頻率超過設置的閾值)，系統服務就會拒絕請求。強行限制系統單位時間內訪問的請求量。漏桶算法示意圖如下：

漏桶算法有兩個關鍵變量：桶的大小和出水速率，他們共同決定了單位時間內系統能接收的最大請求量。因為漏桶算法中桶的大小和出水速率是固定的參數。不能使流突發到端口，對存在突發特性的流量缺乏效率，什么意思呢？我們后邊會使用使用php實現一個漏桶demo，并對測試結果做詳細說明。github源碼地址是：漏桶算法demo

令牌桶(Token Bucket)和漏桶(Leaky Bucket)使用方向相反的算法，這種算法更加容易理解。隨著時間的流逝，系統會按照恒定1/QPS(如果QPS=1000，則時間間隔是1ms)向桶中添加Token(想象和漏洞漏水相反,有個水龍頭在不斷的加水)。如果桶已經滿了就不會添加了，請求到來時會嘗試從桶中拿一個Token,如果拿不到Token,就阻塞或者拒絕服務，待下次有令牌時再去拿令牌。令牌桶的算法如下圖所示例：

令牌桶的好處是顯而易見的，我們可以通過提高放入桶中令牌的速率，改變請求的限制速度。令牌桶一般會定時的向桶中添加令牌(例如每隔10ms向桶中添加一枚令牌)。我們會使用Go語言實現一個令牌桶demo，為了達到兼容分布式并發場景，我們會對令牌桶的demo做改進說明，我們在添加令牌時采用一種變種算法：等請求到達時根據令牌放入桶中的速率實時計算應該放入桶中令牌的數量。github源碼地址是：令牌桶算法demo

我們模擬實現的功能是限制一個公司下對某一個接口的訪問頻次，示例中是限制公司org1的員工列表接口/user/list在1s內能被外部訪問100次。

Redis中設置接口限制1s內訪問100次的hash:

 hmset org1/user/list expire 1 limitReq 100

我們使用Predis連接redis進行操作，模擬接口比較簡單，我們只獲取兩個參數，org和pathInfo,RateLimit類中相關方法是：

 * Date: 2019-06-13
 */

class RateLimit
{
    private $conn = null;       //redis連接
    private $org = "";          //公司標識
    private $pathInfo = "";     //接口路徑信息

    /**
     * RateLimit constructor.
     * @param $org
     * @param $pathInfo
     * @param $expire
     * @param $limitReq
     */
    public function __construct($org, $pathInfo)
    {
        $this->conn = $this->getRedisConn();
        $this->org = $org;
        $this->pathInfo = $pathInfo;
    }
    //......此處省略getLuaScript方法
    /**
     * 獲取redis連接
     * @return PredisClient
     */
    private function getRedisConn()
    {
        require_once("vendor/autoload.php");
        $conn = new PredisClient(["host" => "127.0.0.1",
            "port" => 6379,]);
        return $conn;
    }
    //......此處省略isActionAllowed方法
}

下邊我們看看Lua腳本的設計：

   /**
     * 獲取lua腳本
     * @return string
     */
    private function getLuaScript()
    {
        $luaScript = << tonumber(ARGV[2]) then
return 0
end

return 1
LUA_SCRIPT;

        return $luaScript;
    }

Lua腳本可以打包到Redis服務端進行執行，因為Redis服務端redis-server在2.6版本默認內置了Lua解析器，php的Redis客戶端與Lua腳本交互主要傳兩個KEYS和ARGV,其中KEYS是對應Redis中操作的key值（示例中的KEYS[1]就是org1/user/list),ARGV是要設置的屬性參數。在Lua腳本中Table的索引是從1開始自增的，Lua腳本執行Redis命令可以保證原子性(因為Redis是單線程的)，所以在并發競態條件下也能保證hash的讀寫一致。命令首先調用incr設置org/user/list記數,Redis中的list、set、hash、zset這四種數據結構是容器型數據結構，他們共享下面兩條通用規則：

1.create if not exists：如果容器不存在，那就創建一個再進行操作。比如incr org/user/list時，如果org/user/list不存在，就相當于設置了org/user/list為1,這就是為什么上邊Lua腳本使用expire當times為1時設置org/user/list的過期時間

2.drop if no elements：如果容器里的元素沒有了，那么立即刪除容器，釋放內存。比如lpop操作完一個list之后，list中沒有元素內容了，那么這個list也就不存在了

下邊的邏輯就很明了了，就是看接口的調用累加次數有沒有超限（限制頻率通過ARGV[2]）進行判斷，超限返回0，否則返回1.

下邊我們就可以看看怎樣isActionAllowed方法判斷是否要進行限流了:

    /**
     * 判斷接口是否限制訪問
     * @return bool
     */
    public function isActionAllowed()
    {
        $pathInfo = $this->org . $this->pathInfo;
        $config = $this->conn->hgetall($pathInfo);
        //配置中沒有對接口進行限制
        if (!$config) return true;

        $pathInfoLimitKey = $this->org . "-" . $this->pathInfo;
        try {
            $ret = $this->conn->evalsha(sha1($this->getLuaScript()), 1, $pathInfoLimitKey, $config["expire"], $config["limitReq"]);
        } catch (Exception $e) {
            $ret = $this->conn->eval($this->getLuaScript(), 1, $pathInfoLimitKey, $config["expire"], $config["limitReq"]);
        }

        return boolval($ret);
    }

Predis使用evalsha打包Lua腳本發送到服務端執行。evalsha的第一個參數是sha1編碼后的Lua腳本。redis-server可以對Lua腳本進行緩存，緩存的方法是key:value的形式，其中key是sha1后的lua腳本內容,這樣在Lua腳本比較大時，客戶端只需要發送sha1后的值到redis-server就可以了，減小了每次發送命令內容的字節大小。如果evalsha報出錯誤信息可以改為eval函數，因為redis-server第一次接收到lua腳本，可能還沒沒有進行緩存。最好是使用try...catch...做一下兼容處理。evalsha的第二個參數是key的個數，這里是一個，$pathInfoLimitKey，下邊兩個是從Redis中取出的配置值，標示1s內允許$pathInfoLimitKey被操作100次。如果沒有對$pathInfoLimitKey做配置限制頻率，默認不受限。

以上就是rateLimit類的全部內容了，思路比較簡單，下邊簡單看一下入口文件，也比較簡單，就是接收參數，然后將接口是否受限的信息寫到stat.log日志文件中去。

 * Date: 2019-06-16
 */
require_once("./RateLimit.php");
ini_set("display_errors", true);

$org = $_GET["org"];
$pathInfo = $_GET["path_info"];

$result = (new RateLimit($org, $pathInfo))->isActionAllowed();

$handler = fopen("./stat.log", "a") or die("can not open file!");
if ($result) {
    fwrite($handler, "request success!" . PHP_EOL);
} else {
    fwrite($handler, "request failed!" . PHP_EOL);
}
fclose($handler);

我們通過ab工具壓測一下接口信息，程序限制1s內允許100次訪問，我們就開10個客戶端同時請求110次，理論上應該是前一百次是成功的，后十次是失敗的，命令為:

ab -n 110 -c 10 http://localhost/demo/rateLimit/index.php?org=org1&path_info=/user/list

stat.log中的日志信息和我們預期中的一樣，說明我們的接口頻次設置達到了預期效果：

...//此處省略96行
request success!
request success!
request success!
request success!
request failed!
request failed!
request failed!
request failed!
request failed!
request failed!
request failed!
request failed!
request failed!
request failed!

但是漏斗限流還是有一些缺點的，它不支持突發流量，我們接口設置1s內限制訪問100次，假如說前900毫秒只有80次訪問，突然在接下來的100毫秒來了50次訪問，那么毫無疑問，后邊30次訪問是失敗的。不過漏斗這種簡單粗暴的限流處理方案對于流量集中性訪問，比如(1分鐘只允許訪問1000次)還是非常適合的。

我們首先不考慮競態條件，用go語言實現一個v1版本的令牌桶來體會一下它的算法思想。我們新建一個funnel模塊，定義一個結構體，包含了令牌桶需要的屬性：

package funnel

import (
    "math"
    "time"
)

type Funnel struct {
    Capacity          int64   //令牌桶容量
    LeakingRate       float64 //令牌桶流水速率:每毫秒向令牌桶中添加的令牌數
    RemainingCapacity int64   //令牌桶剩余空間
    LastLeakingTime   int64   //上次流水(放入令牌)時間:毫秒時間戳

Funnel結構體支持導出，分別包含令牌桶的容量、向令牌桶中添加令牌的速率、令牌桶剩余空間
和上次放入令牌時間的四個屬性。
我們采用請求進來時實時改變令牌桶狀態的思路，改變令牌桶狀態的方法如下：

//有請求時更新令牌桶的狀態,主要是令牌桶剩余空間和記錄取走Token的時間戳
func (rateLimit *Funnel) updateFunnelStatus() {
    nowTs := time.Now().UnixNano() / int64(time.Millisecond)
    //距離上一次取走令牌已經過去了多長時間
    timeDiff := nowTs - rateLimit.LastLeakingTime
    //根據時間差和流水速率計算需要向令牌桶中添加多少令牌
    needAddSpace := int64(math.Floor(rateLimit.LeakingRate * float64(timeDiff)))
    //不需要添加令牌
    if needAddSpace < 1 {
        return
    }
    rateLimit.RemainingCapacity += needAddSpace
    //添加的令牌不能大于令牌桶的剩余空間
    if rateLimit.RemainingCapacity > rateLimit.Capacity {
        rateLimit.RemainingCapacity = rateLimit.Capacity
    }
    //更新上次令牌桶流水(添加令牌)時間戳
    rateLimit.LastLeakingTime = nowTs
}

因為要改變令牌桶的狀態，所以我們這里使用指針接收者為結構體Funnel定義方法。主要思路就是根據當前時間和上次放入令牌桶中令牌的時間戳，再結合每毫秒應該放入令牌桶中令牌，計算添加應該放入到令牌桶中的令牌，放入令牌后不能超過令牌桶本身容量的大小。然后取出令牌，更新上次添加令牌時間戳。
判斷接口是否限流其實就是看能不能從令牌桶中取出令牌，方法如下：

//判斷接口是否被限流
func (rateLimit *Funnel) IsActionAllowed() bool {
    //更新令牌桶狀態
    rateLimit.updateFunnelStatus()
    if rateLimit.RemainingCapacity < 1 {
        return false
    }
    rateLimit.RemainingCapacity = rateLimit.RemainingCapacity - 1
    return true
}

到了這里，相信讀者已經對令牌桶算法有了一個比較清晰的認識了。我們再來說問題，因為限流最終還是要通過操作Redis來實現的，我們首先來在Redis里初始化好接口限流的配置：

hmset org2/user/list Capacity 100 LeakingRate 0.1 RemainingCapacity 0 LastLeakingTime 1560789716896

我們設置公司二(org2)的接口（/user/list）令牌桶容量100，每隔10ms放入一令牌（計算方法100/1000）。我們將Funnel對象內容的字段存儲到一個hash結構中，我們在計算是否限流的時候需要從hash結構中取值，在內存中做運算，再回填到hash結構，尤其對于go語言這種天然并發的程序來講，我們無法保證整個過程的原子化（這就是為什么要使用Lua腳本的原因，因為如果用程序來實現，就需要加鎖，一旦加鎖就有加鎖失敗的可能，失敗只能選擇重試或放棄，重試會導致性能下降，放棄會影響用戶體驗，代碼復雜度會增加不少）。我們V2版本還是會選擇使用Lua腳本來實現：具體調研過程如下：

補充一點：Redis4.0提供了一個限流模塊Redis模塊，它叫Redis-Cell。該模塊也使用了漏斗算法，并提供了原子的限流命令，重試機制也非常簡單，有興趣的可以研究一下。我們這里還是使用Lua + Redis解決方案，廢話少說，上V2版本的代碼：

const luaScript = `
-- 接口限流
-- last_leaking_time 最后訪問時間的毫秒
-- remaining_capacity 當前令牌桶中可用請求令牌數
-- capacity 令牌桶容量
-- leaking_rate    令牌桶添加令牌的速率

-- 把發生數據變更的命令以事務的方式做持久化和主從復制(Redis4.0支持)
redis.replicate_commands()

-- 獲取令牌桶的配置信息
local rate_limit_info = redis.call("HGETALL", KEYS[1])

-- 獲取當前時間戳
local timestamp = redis.call("TIME")
local now = math.floor((timestamp[1] * 1000000 + timestamp[2]) / 1000)

if rate_limit_info == nil then -- 沒有設置限流配置,則默認拿到令牌
    return now * 10 + 1
end

local capacity = tonumber(rate_limit_info[2])
local leaking_rate = tonumber(rate_limit_info[4])
local remaining_capacity = tonumber(rate_limit_info[6])
local last_leaking_time = tonumber(rate_limit_info[8])

-- 計算需要補給的令牌數,更新令牌數和補給時間戳
local supply_token = math.floor((now - last_leaking_time) * leaking_rate)
if (supply_token > 0) then
   last_leaking_time = now
   remaining_capacity = supply_token + remaining_capacity
   if remaining_capacity > capacity then
      remaining_capacity = capacity
   end
end

local result = 0 -- 返回結果是否能夠拿到令牌,默認否

-- 計算請求是否能夠拿到令牌
if (remaining_capacity > 0) then
    remaining_capacity = remaining_capacity - 1
    result = 1
end

-- 更新令牌桶的配置信息
redis.call("HMSET", KEYS[1], "RemainingCapacity", remaining_capacity, "LastLeakingTime", last_leaking_time)

return now * 10 + result
`

我們這段腳本返回一個int64類型的整數，最后一位0或1表示是否要對接口限流，前邊的數字表示毫秒時間戳，將來記錄到日志里進行壓測統計使用。程序運行時當前時間戳我是調用Redis的time命令計算獲得的，原因有二：

Lua命令獲得當前時間戳只能精確到秒，而Redis確可以精確到納秒。

如果時間戳作為腳本調用參數（go程序）傳進來會有問題，因為腳本傳參到Lua在Redis中執行還有一段時間誤差，不能保證最先被接收到的請求先被處理，而Lua中獲取時間戳可以保證請求、時間串行

和以前一樣，沒有設置限流配置，就默認可以請求。
然后根據時間戳補給令牌，計算是否能夠取到令牌，然后更新令牌狀態，思路和V1版本一樣，讀者可自行閱讀。說明一點，腳本開始處的redis.replicate_commands()命令是因為Redis低版本不支持對Redis既讀又寫，所以這種方式還是存在版本兼容性，但是解決辦法確是最完美的。
接下來我們看go邏輯代碼：

func main() {
    http.HandleFunc("/user/list", handleReq)
    http.ListenAndServe(":8082", nil)
}

//初始化redis連接池
func newPool() *redis.Pool {
    return &redis.Pool{
        MaxIdle:   80,
        MaxActive: 12000, // max number of connections
        Dial: func() (redis.Conn, error) {
            c, err := redis.Dial("tcp", ":6379")
            if err != nil {
                panic(err.Error())
            }
            return c, err
        },
    }
}

//寫入日志
func writeLog(msg string, logPath string) {
    fd, _ := os.OpenFile(logPath, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
    defer fd.Close()
    content := strings.Join([]string{msg, "
"}, "")
    buf := []byte(content)
    fd.Write(buf)
}

//處理請求函數,根據請求將響應結果信息寫入日志
func handleReq(w http.ResponseWriter, r *http.Request) {
    //獲取url信息
    pathInfo := r.URL.Path
    //獲取get傳遞的公司信息org
    orgInfo, ok := r.URL.Query()["org"]
    if !ok || len(orgInfo) < 1 {
        fmt.Println("Param org is missing!")
    }

    //調用lua腳本原子性進行接口限流統計
    conn := newPool().Get()
    key := orgInfo[0] + pathInfo
    lua := redis.NewScript(1, luaScript)
    reply, err := redis.Int64(lua.Do(conn, key))
    if err != nil {
        fmt.Println(err)
        return
    }
    //接口是否被限制訪問
    isLimit := bool(reply % 10 == 1)
    reqTime := int64(math.Floor(float64(reply) / 10))
    //將統計結果寫入日志當中
    if !isLimit {
        successLog := strconv.FormatInt(reqTime, 10) + " request failed!"
        writeLog(successLog, "./stat.log")
        return
    }

    failedLog := strconv.FormatInt(reqTime, 10) + " request success!"
    writeLog(failedLog, "./stat.log")
}

腳本監聽本地8082端口，使用go的redis框架redigo來操作redis，我們初始化了一個redis連接池，從連接池中取得連接進行操作。我們分析如下代碼：

lua := redis.NewScript(1, luaScript)
    reply, err := redis.Int64(lua.Do(conn, key))

NewScript中第一個參數代表要操作Redis的key的個數，這點和Predis的evalsha第二個參數類似。然后采用Do方法執行腳本，返回值使用redis.Int64做處理，然后進行運算判斷接口是否允許被訪問，然后將訪問時間和結果寫入到stat.log日志文件中。
邏輯還是非常的簡單，我們主要看壓測結果，啟動代碼，使用ab壓測命令執行：

 ab -n 110 -c 10 http://127.0.0.1:8082/user/list?org=org2

然后我們分析stat.log日志興許會有些驚訝：

1561263349294 request success!    //第一行日志
...//省略95行
1561263349387 request success!
1561263349388 request success!
1561263349398 request success!
1561263349396 request success!
1561263349404 request success!
1561263349407 request success!
1561263349406 request success!
1561263349406 request success!
1561263349407 request success!
1561263349406 request success!
1561263349406 request success!
1561263349405 request success!
1561263349406 request success!
1561263349406 request success!
1561263349406 request success!

是的，都成功了，為什么呢？我們看統計時間會發現執行這100個請求總共用了110毫秒，在程序執行過程中，每隔10ms會向令牌桶中添加一個令牌，一共添加了11個令牌，所以110次請求都拿到了令牌。可以看出令牌桶適用于大流量下的限流，可以保證流量按照時間均勻分攤，避免出現流量的集中式爆發訪問。

到此為止，已經給大家介紹了限流的必要性以及常用限流手段與程序實現。相信大家對分步式限流有了一個初步的了解。下面做一個簡單的總結: