Redis + Lua 接口限流最佳實踐策略

1.應用場景

咱們開發的接口服務系統不少都具備抗高併發，保證高可用的特性。現實條件下，隨着流量的不斷增長，在經費、硬件和資源受限的狀況下，咱們就須要爲咱們的系統服務制定有效的限流、分流策略來保護咱們的系統了。

2.算法簡介和示例說明

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

2.1漏桶算法

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

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

2.2令牌桶算法

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

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

2.3示例說明

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

3.示例源碼和壓測結果

3.1 php實現漏桶算法

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

hmset org1/user/list expire 1 limitReq 100

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

<?php
/**
 * Description: 漏桶限流
 * User: guozhaoran<guozhaoran@cmcm.com>
 * 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 \Predis\Client
     */
    private function getRedisConn()
    {
        require_once('vendor/autoload.php');
        $conn = new Predis\Client(['host' => '127.0.0.1',
            'port' => 6379,]);
        return $conn;
    }
    //......此處省略isActionAllowed方法
}

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

/**
     * 獲取lua腳本
     * @return string
     */
    private function getLuaScript()
    {
        $luaScript = <<<LUA_SCRIPT
-- 限制接口訪問頻次
local times = redis.call('incr', KEYS[1]);    --將key自增1

if times == 1 then
redis.call('expire', KEYS[1], ARGV[1])    --給key設置過時時間
end

if times > 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日誌文件中去。

<?php
/**
 * Description: 漏斗限流入口文件
 * User: guozhaoran<guozhaoran@cmcm.com>
 * 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次)仍是很是適合的。

3.2 go語言實現令牌桶算法

咱們首先不考慮競態條件，用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腳原本實現：具體調研過程以下：

方案	特色
單服務對操做採用鎖機制	文章有提到，這種只能保證單節點下串行且性能差
Redis原子操做incr	這種方案咱們在漏斗模型中有使用，它只能應對簡單的場景，涉及到複雜場景就比較難處理
Redis分佈式事務	雖然Redis的分佈式事務能保證原子操做，可是實現複雜，而且網絡開銷大，須要大量的網絡傳輸
Redis+Lua	這裏就不得不誇一下這種方案了，Lua腳本中運行在Redis中，redis又是單線程的，所以能保證操做的串行。另外：減小網絡開銷，前邊咱們提到過，Lua代碼包裝的命令不須要發送屢次命令請求，Redis能夠對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, "\r\n"}, "")
    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次請求都拿到了令牌。能夠看出令牌桶適用於大流量下的限流，能夠保證流量按照時間均勻分攤，避免出現流量的集中式爆發訪問。

4.簡單總結

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

算法	場景
令牌桶	適用於大流量下的訪問，能夠保證流量按時間均勻分攤，避免出現流量集中爆發式訪問
漏桶	簡單粗暴，對於大流量下限流有很好的效果，尤爲適合於單位時間內限制請求的業務，對突發流量的不能有很好的應對