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脚本来实现：具体调研过程如下：

补充一点：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.简单总结

到此为止，已经给大家介绍了限流的必要性以及常用限流手段与程序实现。相信大家对分步式限流有了一个初步的了解。下面做一个简单的总结:

有疑问加站长微信联系（非本文作者）