摘要：算法簡(jiǎn)介和示例說(shuō)明業(yè)界比較流行的限流算法有漏桶算法和令牌桶算法。判斷接口是否限流其實(shí)就是看能不能從令牌桶中取出令牌，方法如下判斷接口是否被限流更新令牌桶狀態(tài)到了這里，相信讀者已經(jīng)對(duì)令牌桶算法有了一個(gè)比較清晰的認(rèn)識(shí)了。

我們開(kāi)發(fā)的接口服務(wù)系統(tǒng)很多都具有抗高并發(fā)，保證高可用的特性。現(xiàn)實(shí)條件下，隨著流量的不斷增加，在經(jīng)費(fèi)、硬件和資源受限的情況下，我們就需要為我們的系統(tǒng)服務(wù)制定有效的限流、分流策略來(lái)保護(hù)我們的系統(tǒng)了。

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

漏桶(Leaky Bucket)算法的實(shí)現(xiàn)思路比較簡(jiǎn)單，水(請(qǐng)求)先流入到桶中，然后桶以一定的速度出水(接口有響應(yīng)速率)，當(dāng)水流過(guò)大時(shí)(訪問(wèn)頻率超過(guò)設(shè)置的閾值)，系統(tǒng)服務(wù)就會(huì)拒絕請(qǐng)求。強(qiáng)行限制系統(tǒng)單位時(shí)間內(nèi)訪問(wèn)的請(qǐng)求量。漏桶算法示意圖如下：

漏桶算法有兩個(gè)關(guān)鍵變量：桶的大小和出水速率，他們共同決定了單位時(shí)間內(nèi)系統(tǒng)能接收的最大請(qǐng)求量。因?yàn)槁┩八惴ㄖ型暗拇笮『统鏊俾适枪潭ǖ膮?shù)。不能使流突發(fā)到端口，對(duì)存在突發(fā)特性的流量缺乏效率，什么意思呢？我們后邊會(huì)使用使用php實(shí)現(xiàn)一個(gè)漏桶demo，并對(duì)測(cè)試結(jié)果做詳細(xì)說(shuō)明。github源碼地址是：漏桶算法demo

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

令牌桶的好處是顯而易見(jiàn)的，我們可以通過(guò)提高放入桶中令牌的速率，改變請(qǐng)求的限制速度。令牌桶一般會(huì)定時(shí)的向桶中添加令牌(例如每隔10ms向桶中添加一枚令牌)。我們會(huì)使用Go語(yǔ)言實(shí)現(xiàn)一個(gè)令牌桶demo，為了達(dá)到兼容分布式并發(fā)場(chǎng)景，我們會(huì)對(duì)令牌桶的demo做改進(jìn)說(shuō)明，我們?cè)谔砑恿钆茣r(shí)采用一種變種算法：等請(qǐng)求到達(dá)時(shí)根據(jù)令牌放入桶中的速率實(shí)時(shí)計(jì)算應(yīng)該放入桶中令牌的數(shù)量。github源碼地址是：令牌桶算法demo

我們模擬實(shí)現(xiàn)的功能是限制一個(gè)公司下對(duì)某一個(gè)接口的訪問(wèn)頻次，示例中是限制公司org1的員工列表接口/user/list在1s內(nèi)能被外部訪問(wèn)100次。

Redis中設(shè)置接口限制1s內(nèi)訪問(wèn)100次的hash:

 hmset org1/user/list expire 1 limitReq 100

我們使用Predis連接redis進(jìn)行操作，模擬接口比較簡(jiǎn)單，我們只獲取兩個(gè)參數(shù)，org和pathInfo,RateLimit類中相關(guān)方法是：

 * Date: 2019-06-13
 */

class RateLimit
{
    private $conn = null;       //redis連接
    private $org = "";          //公司標(biāo)識(shí)
    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腳本的設(shè)計(jì)：

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

return 1
LUA_SCRIPT;

        return $luaScript;
    }

Lua腳本可以打包到Redis服務(wù)端進(jìn)行執(zhí)行，因?yàn)镽edis服務(wù)端redis-server在2.6版本默認(rèn)內(nèi)置了Lua解析器，php的Redis客戶端與Lua腳本交互主要傳兩個(gè)KEYS和ARGV,其中KEYS是對(duì)應(yīng)Redis中操作的key值（示例中的KEYS[1]就是org1/user/list),ARGV是要設(shè)置的屬性參數(shù)。在Lua腳本中Table的索引是從1開(kāi)始自增的，Lua腳本執(zhí)行Redis命令可以保證原子性(因?yàn)镽edis是單線程的)，所以在并發(fā)競(jìng)態(tài)條件下也能保證hash的讀寫(xiě)一致。命令首先調(diào)用incr設(shè)置org/user/list記數(shù),Redis中的list、set、hash、zset這四種數(shù)據(jù)結(jié)構(gòu)是容器型數(shù)據(jù)結(jié)構(gòu)，他們共享下面兩條通用規(guī)則：

1.create if not exists：如果容器不存在，那就創(chuàng)建一個(gè)再進(jìn)行操作。比如incr org/user/list時(shí)，如果org/user/list不存在，就相當(dāng)于設(shè)置了org/user/list為1,這就是為什么上邊Lua腳本使用expire當(dāng)times為1時(shí)設(shè)置org/user/list的過(guò)期時(shí)間

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

下邊的邏輯就很明了了，就是看接口的調(diào)用累加次數(shù)有沒(méi)有超限（限制頻率通過(guò)ARGV[2]）進(jìn)行判斷，超限返回0，否則返回1.

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

    /**
     * 判斷接口是否限制訪問(wèn)
     * @return bool
     */
    public function isActionAllowed()
    {
        $pathInfo = $this->org . $this->pathInfo;
        $config = $this->conn->hgetall($pathInfo);
        //配置中沒(méi)有對(duì)接口進(jìn)行限制
        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腳本發(fā)送到服務(wù)端執(zhí)行。evalsha的第一個(gè)參數(shù)是sha1編碼后的Lua腳本。redis-server可以對(duì)Lua腳本進(jìn)行緩存，緩存的方法是key:value的形式，其中key是sha1后的lua腳本內(nèi)容,這樣在Lua腳本比較大時(shí)，客戶端只需要發(fā)送sha1后的值到redis-server就可以了，減小了每次發(fā)送命令內(nèi)容的字節(jié)大小。如果evalsha報(bào)出錯(cuò)誤信息可以改為eval函數(shù)，因?yàn)閞edis-server第一次接收到lua腳本，可能還沒(méi)沒(méi)有進(jìn)行緩存。最好是使用try...catch...做一下兼容處理。evalsha的第二個(gè)參數(shù)是key的個(gè)數(shù)，這里是一個(gè)，$pathInfoLimitKey，下邊兩個(gè)是從Redis中取出的配置值，標(biāo)示1s內(nèi)允許$pathInfoLimitKey被操作100次。如果沒(méi)有對(duì)$pathInfoLimitKey做配置限制頻率，默認(rèn)不受限。

以上就是rateLimit類的全部?jī)?nèi)容了，思路比較簡(jiǎn)單，下邊簡(jiǎn)單看一下入口文件，也比較簡(jiǎn)單，就是接收參數(shù)，然后將接口是否受限的信息寫(xiě)到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);

我們通過(guò)ab工具壓測(cè)一下接口信息，程序限制1s內(nèi)允許100次訪問(wèn)，我們就開(kāi)10個(gè)客戶端同時(shí)請(qǐng)求110次，理論上應(yīng)該是前一百次是成功的，后十次是失敗的，命令為:

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

stat.log中的日志信息和我們預(yù)期中的一樣，說(shuō)明我們的接口頻次設(shè)置達(dá)到了預(yù)期效果：

...//此處省略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!

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

我們首先不考慮競(jìng)態(tài)條件，用go語(yǔ)言實(shí)現(xiàn)一個(gè)v1版本的令牌桶來(lái)體會(huì)一下它的算法思想。我們新建一個(gè)funnel模塊，定義一個(gè)結(jié)構(gòu)體，包含了令牌桶需要的屬性：

package funnel

import (
    "math"
    "time"
)

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

Funnel結(jié)構(gòu)體支持導(dǎo)出，分別包含令牌桶的容量、向令牌桶中添加令牌的速率、令牌桶剩余空間
和上次放入令牌時(shí)間的四個(gè)屬性。
我們采用請(qǐng)求進(jìn)來(lái)時(shí)實(shí)時(shí)改變令牌桶狀態(tài)的思路，改變令牌桶狀態(tài)的方法如下：

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

因?yàn)橐淖兞钆仆暗臓顟B(tài)，所以我們這里使用指針接收者為結(jié)構(gòu)體Funnel定義方法。主要思路就是根據(jù)當(dāng)前時(shí)間和上次放入令牌桶中令牌的時(shí)間戳，再結(jié)合每毫秒應(yīng)該放入令牌桶中令牌，計(jì)算添加應(yīng)該放入到令牌桶中的令牌，放入令牌后不能超過(guò)令牌桶本身容量的大小。然后取出令牌，更新上次添加令牌時(shí)間戳。
判斷接口是否限流其實(shí)就是看能不能從令牌桶中取出令牌，方法如下：

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

到了這里，相信讀者已經(jīng)對(duì)令牌桶算法有了一個(gè)比較清晰的認(rèn)識(shí)了。我們?cè)賮?lái)說(shuō)問(wèn)題，因?yàn)橄蘖髯罱K還是要通過(guò)操作Redis來(lái)實(shí)現(xiàn)的，我們首先來(lái)在Redis里初始化好接口限流的配置：

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

我們?cè)O(shè)置公司二(org2)的接口（/user/list）令牌桶容量100，每隔10ms放入一令牌（計(jì)算方法100/1000）。我們將Funnel對(duì)象內(nèi)容的字段存儲(chǔ)到一個(gè)hash結(jié)構(gòu)中，我們?cè)谟?jì)算是否限流的時(shí)候需要從hash結(jié)構(gòu)中取值，在內(nèi)存中做運(yùn)算，再回填到hash結(jié)構(gòu)，尤其對(duì)于go語(yǔ)言這種天然并發(fā)的程序來(lái)講，我們無(wú)法保證整個(gè)過(guò)程的原子化（這就是為什么要使用Lua腳本的原因，因?yàn)槿绻贸绦騺?lái)實(shí)現(xiàn)，就需要加鎖，一旦加鎖就有加鎖失敗的可能，失敗只能選擇重試或放棄，重試會(huì)導(dǎo)致性能下降，放棄會(huì)影響用戶體驗(yàn)，代碼復(fù)雜度會(huì)增加不少）。我們V2版本還是會(huì)選擇使用Lua腳本來(lái)實(shí)現(xiàn)：具體調(diào)研過(guò)程如下：

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

const luaScript = `
-- 接口限流
-- last_leaking_time 最后訪問(wèn)時(shí)間的毫秒
-- remaining_capacity 當(dāng)前令牌桶中可用請(qǐng)求令牌數(shù)
-- capacity 令牌桶容量
-- leaking_rate    令牌桶添加令牌的速率

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

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

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

if rate_limit_info == nil then -- 沒(méi)有設(shè)置限流配置,則默認(rèn)拿到令牌
    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])

-- 計(jì)算需要補(bǔ)給的令牌數(shù),更新令牌數(shù)和補(bǔ)給時(shí)間戳
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 -- 返回結(jié)果是否能夠拿到令牌,默認(rèn)否

-- 計(jì)算請(qǐng)求是否能夠拿到令牌
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
`

我們這段腳本返回一個(gè)int64類型的整數(shù)，最后一位0或1表示是否要對(duì)接口限流，前邊的數(shù)字表示毫秒時(shí)間戳，將來(lái)記錄到日志里進(jìn)行壓測(cè)統(tǒng)計(jì)使用。程序運(yùn)行時(shí)當(dāng)前時(shí)間戳我是調(diào)用Redis的time命令計(jì)算獲得的，原因有二：

Lua命令獲得當(dāng)前時(shí)間戳只能精確到秒，而Redis確可以精確到納秒。

如果時(shí)間戳作為腳本調(diào)用參數(shù)（go程序）傳進(jìn)來(lái)會(huì)有問(wèn)題，因?yàn)槟_本傳參到Lua在Redis中執(zhí)行還有一段時(shí)間誤差，不能保證最先被接收到的請(qǐng)求先被處理，而Lua中獲取時(shí)間戳可以保證請(qǐng)求、時(shí)間串行

和以前一樣，沒(méi)有設(shè)置限流配置，就默認(rèn)可以請(qǐng)求。
然后根據(jù)時(shí)間戳補(bǔ)給令牌，計(jì)算是否能夠取到令牌，然后更新令牌狀態(tài)，思路和V1版本一樣，讀者可自行閱讀。說(shuō)明一點(diǎn)，腳本開(kāi)始處的redis.replicate_commands()命令是因?yàn)镽edis低版本不支持對(duì)Redis既讀又寫(xiě)，所以這種方式還是存在版本兼容性，但是解決辦法確是最完美的。
接下來(lái)我們看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
        },
    }
}

//寫(xiě)入日志
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)
}

//處理請(qǐng)求函數(shù),根據(jù)請(qǐng)求將響應(yīng)結(jié)果信息寫(xiě)入日志
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!")
    }

    //調(diào)用lua腳本原子性進(jìn)行接口限流統(tǒng)計(jì)
    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
    }
    //接口是否被限制訪問(wèn)
    isLimit := bool(reply % 10 == 1)
    reqTime := int64(math.Floor(float64(reply) / 10))
    //將統(tǒng)計(jì)結(jié)果寫(xiě)入日志當(dāng)中
    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")
}

腳本監(jiān)聽(tīng)本地8082端口，使用go的redis框架redigo來(lái)操作redis，我們初始化了一個(gè)redis連接池，從連接池中取得連接進(jìn)行操作。我們分析如下代碼：

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

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

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

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

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!

是的，都成功了，為什么呢？我們看統(tǒng)計(jì)時(shí)間會(huì)發(fā)現(xiàn)執(zhí)行這100個(gè)請(qǐng)求總共用了110毫秒，在程序執(zhí)行過(guò)程中，每隔10ms會(huì)向令牌桶中添加一個(gè)令牌，一共添加了11個(gè)令牌，所以110次請(qǐng)求都拿到了令牌?？梢钥闯隽钆仆斑m用于大流量下的限流，可以保證流量按照時(shí)間均勻分?jǐn)?，避免出現(xiàn)流量的集中式爆發(fā)訪問(wèn)。

到此為止，已經(jīng)給大家介紹了限流的必要性以及常用限流手段與程序?qū)崿F(xiàn)。相信大家對(duì)分步式限流有了一個(gè)初步的了解。下面做一個(gè)簡(jiǎn)單的總結(jié):