PHP+Redis实现限流

发表于   |   分类于   |  

在一个分布式的高可用系统中，限流是必备的操作。这个流可以是：网络流量，带宽，每秒处理的事务数，每秒请求数，并发请求数，或者业务上的指标等。

找了很多资料,写的最详细的是王争老师的一篇文章 微服务接口限流的设计与思考 。所以决定给公司的API优化下，加入接口的限流！因为公司的项目都是Laravel写的，所以把限流的操作都放到了中间件中，这样可以统一配置。

下面的实现都是通过使用Redis来写的Laravel中间件。

创建一个中间件的话可以自己手动创建，也可以通过命令 php artisan make:middleware xxx。如果想要全部路由都可以使用 加入到 app/Http/Kernel.php 的 $middleware 数组中即可。 如果想要自己指定部分路由使用中间件就加入到 $routeMiddleware 数组中，然后显示指定路由调用。

另外我还简单写了个模拟的配置文件如下 config/grant.php：

1
2
3
4
return [
    'limit' => 3, // 请求限制量
    'interval' => 10000, // 请求单位时间 毫秒
];

固定时间窗口限流

这种也叫计数器法限流，是限流算法里最简单也是最容易实现的一种算法。主要通过来一个记一个，然后判断在有限时间窗口内的数量是否超过限制即可。

比如限制每秒100次请求，多余的请求就拒绝或者阻塞排队。

但是这会出现一个问题，比如第一秒的最后10ms,和第二秒的前10ms内有大量的请求同时到来，根据计算他们是合法的其结果是 >100 所以有可能压垮系统。

php实现伪代码： GrantMiddleware.php

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
<?php
namespace App\Http\Middleware;
use Closure;
use Illuminate\Support\Facades\Log;
use Predis\Client;
class GrantMiddleware
{
    /**
     * redis版计数器限流
     * 这个是直接给每个接口限流，还可以改进成根据token来对人实现限流
     * 比如从token中解析出用户id，然后改成对每个用户请求接口进行限流
     *
     * @param  \Illuminate\Http\Request  $request
     * @param  \Closure  $next
     * @return mixed
     */
    public function handle($request, Closure $next)
    {
        // 获取路由
        $url = $request->getRequestUri();
        $questionLocation = strpos($url, '?');
        $url = substr($url, 0, $questionLocation);
        $result = $this->grant($url);
        if (! $result) {
            return response()->json(['status' => false, 'message' => 'network busy, please try again', 'code' => 429, 'data' => []]);
        }
        return $next($request);
    }
    /**
     * 限流鉴权-基于redis的hash
     *
     * @param $url
     * @return bool
     */
    public function grant($url)
    {
        try {
            $redis = new Client();
            $limit = config('grant.limit');
            $interval = (int)config('grant.interval');
            $check = $redis->hLen($url);
            if (! $check) {
                return $this->initGrant($redis, $url);
            }
            $now = $this->getMicroSecond();
            $requestCount = $redis->hGet($url, 'requestCount');
            $timestamp = (int)$redis->hGet($url, 'timestamp');
            if ($now < $timestamp + $interval) {
                $requestCount++;
                return $requestCount <= $limit;
            } else {
                return $this->initGrant($redis, $url);
            }
        } catch (\Exception $exception) {
            Log::info('GrantMiddleware|'.date('Y-m-d H:i:s', time()).'|grant()方法捕获异常'.'|'.
                $exception->getFile().'|'.$exception->getCode().'|'.$exception->getMessage());
            return false;
        }
    }
    /**
     * 初始化指定路由的限流
     *
     * @param $redis
     * @param $url
     * @return bool
     */
    public function initGrant($redis, $url)
    {
        $redis->hSet($url, 'timestamp', $this->getMicroSecond());
        $redis->hSet($url, 'requestCount', 1);
        return true;
    }
    /**
     * 重新计算时间戳 毫秒
     *
     * @return float
     */
    public function getMicroSecond()
    {
        list($microsecond, $second) = explode(' ', microtime());
        $microsecond = (float)sprintf('%.0f', (floatval($microsecond) + floatval($second)) * 1000);
        return $microsecond;
    }
}

滑动时间窗口限流

滑动时间窗口限流是固定时间窗口的优化版，可以解决它的2个时间交汇点并发过高问题。

对比固定时间窗口限流算法，滑动时间窗口限流算法的时间窗口是持续滑动的，并且除了需要一个计数器来记录时间窗口内接口请求次数之外，还需要记录在时间窗口内每个接口请求到达的时间点，对内存的占用会比较多。

实现步骤：

滑动窗口记录的时间点 list = (t_1, t_2, …t_k)，时间窗口大小为 1 秒，起点是 list 中最小的时间点。当 t_m 时刻新的请求到来时，我们通过以下步骤来更新滑动时间窗口并判断是否限流熔断：

Step1: 检查接口请求的时间 t_m 是否在当前的时间窗口 [t_start, t_start+1 秒) 内。如果是，则跳转到 STEP 3，否则跳转到 STEP 2.

Step2: 向后滑动时间窗口，将时间窗口的起点 t_start 更新为 list 中的第二小时间点，并将最小的时间点从 list 中删除。然后，跳转到 STEP 1。

Step3: 判断当前时间窗口内的接口请求数是否小于最大允许的接口请求限流值，即判断: list.size < max_hits_limit，如果小于，则说明没有超过限流值，允许接口请求，并将此接口请求的访问时间放入到时间窗口内，否则直接执行限流熔断。

滑动时间窗口

即便滑动时间窗口限流算法可以保证任意时间窗口内接口请求次数都不会超过最大限流值，但是仍然不能防止在细时间粒度上面访问过于集中的问题，基于时间窗口的限流算法，不管是固定时间窗口还是滑动时间窗口，只能在选定的时间粒度上限流，对选定时间粒度内的更加细粒度的访问频率不做限制。比如最后1ns内200次请求系统还是要挂！

php实现伪代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
<?php
namespace App\Http\Middleware;
use Closure;
use Predis\Client;
use Illuminate\Support\Facades\Log;
class SlidingWindowGrantMiddleware
{
    /**
     * 基于redis的滑动窗口限流
     *
     * @param  \Illuminate\Http\Request  $request
     * @param  \Closure  $next
     * @return mixed
     */
    public function handle($request, Closure $next)
    {
        // 获取路由
        $url = $request->getRequestUri();
        $questionLocation = strpos($url, '?');
        $url = 'window:'.substr($url, 0, $questionLocation);
        $result = $this->slidingWindowGrant($url);
        if (! $result) {
            return response()->json(['status' => false, 'message' => 'network busy, please try again', 'code' => 429, 'data' => []]);
        }
        return $next($request);
    }
    /**
     * 限流鉴权-基于redis的list
     *
     * @param $url
     * @return boolean
     */
    public function slidingWindowGrant($url)
    {
        try {
            $redis = new Client();
            $limit = config('grant.limit');
            $interval = (int)config('grant.interval');
            $size = (int)$redis->lLen($url);
            if (! $size) {
                return $this->insertSlidingWindow($redis, $url, $this->getMicroSecond());
            }
            $now = $this->getMicroSecond();
            $startTime = (float)$redis->lindex($url, 0);
            $check = $size;
            while ($check > 0) {
                if ($now >= $startTime + $interval) {
                    $startTime = (int)$redis->lpop($url);
                    $check--;
                    continue;
                } else {
                    break;
                }
            }
            if ($now <= $startTime + $interval) {
                if ($size < $limit) {
                    return $this->insertSlidingWindow($redis, $url, $now);
                } else {
                    return false;
                }
            } else {
                return $this->insertSlidingWindow($redis, $url, $now);
            }
        } catch (\Exception $exception) {
            Log::info('SlidingWindowGrantMiddleware|'.date('Y-m-d H:i:s', time()).'|slidingWindowGrant()方法捕获异常'.'|'.
                $exception->getFile().'|'.$exception->getCode().'|'.$exception->getMessage());
            return false;
        }
    }
    /**
     * 初始化滑动窗口
     *
     * @param $redis
     * @param $url
     * @param $timestamp
     * @return boolean
     */
    public function insertSlidingWindow($redis, $url, $timestamp)
    {
        $redis->rPush($url, $timestamp);
        return true;
    }
    /**
     * 重新计算时间戳 毫秒
     *
     * @return float
     */
    public function getMicroSecond()
    {
        list($microsecond, $second) = explode(' ', microtime());
        $microsecond = (float)sprintf('%.0f', (floatval($microsecond) + floatval($second)) * 1000);
        return $microsecond;
    }
}

令牌桶限流

令牌桶：

  • 接口限制 t 秒内最大访问次数为 n，则每隔 t/n 秒会放一个 token 到桶中；
  • 桶中最多可以存放 b 个 token，如果 token 到达时令牌桶已经满了，那么这个 token 会被丢弃；
  • 接口请求会先从令牌桶中取 token，拿到 token 则处理接口请求，拿不到 token 则执行限流。

令牌桶算法看似比较复杂，每间隔固定时间都要放 token 到桶中，但并不需要专门起一个线程来做这件事情。每次在取 token 之前，根据上次放入 token 的时间戳和现在的时间戳，计算出这段时间需要放多少 token 进去，一次性放进去，所以在实现上面也并没有太大难度。这种的话是王争老师说的，不过我稍微改了下，改成了使用计数器的记录时间然后配合redis的队列。

php实现伪代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
<?php
namespace App\Http\Middleware;
use Closure;
use Predis\Client;
use Illuminate\Support\Facades\Log;
class TokenBucketMiddleware
{
    /**
     * 基于redis的令牌桶限流- 使用计数器的记录时间然后配合redis的队列
     *
     * @param  \Illuminate\Http\Request  $request
     * @param  \Closure  $next
     * @return mixed
     */
    public function handle($request, Closure $next)
    {
        // 获取路由
        $url = $request->getRequestUri();
        $questionLocation = strpos($url, '?');
        $url = 'bucket:'.substr($url, 0, $questionLocation);
        $result = $this->grant($url);
        if (! $result) {
            return response()->json(['status' => false, 'message' => 'network busy, please try again', 'code' => 429, 'data' => []]);
        }
        return $next($request);
    }
    /**
     * 限流鉴权
     *
     * @param $url
     * @return bool
     */
    public function grant($url)
    {
        try {
            $redis = new Client();
            $limit = config('grant.limit');
            $interval = (int)config('grant.interval');
            $timeKey = $url.':time';
            $check = $redis->hLen($timeKey);
            if (! $check) {
                return $this->init($redis, $timeKey, $url, $limit - 1);
            }
            $now = $this->getMicroSecond();
            $timestamp = (int)$redis->hGet($timeKey, 'timestamp');
            if ($now < $timestamp + $interval) {
                // 在时间内
                return $this->getToken($redis, $url);
            } else {
                // 不在时间内
                return $this->init($redis, $timeKey, $url, $limit);
            }
        } catch (\Exception $exception) {
            Log::info('TokenBucketMiddleware|'.date('Y-m-d H:i:s', time()).'|grant()方法捕获异常'.'|'.
                $exception->getFile().'|'.$exception->getCode().'|'.$exception->getMessage());
            return false;
        }
    }
    /**
     * 初始化
     *
     * @param $redis
     * @param $timeKey
     * @param $url
     * @param $limit
     * @return bool
     */
    public function init($redis, $timeKey, $url, $limit)
    {
        // 初始化时间
        $this->initGrant($redis, $timeKey);
        // 初始化令牌桶
        $this->initTokenBucket($redis, $url, $limit);
        return true;
    }
    /**
     * 初始化指定路由的限流
     *
     * @param $redis
     * @param $url
     * @return bool
     */
    public function initGrant($redis, $url)
    {
        $redis->hSet($url, 'timestamp', $this->getMicroSecond());
        return true;
    }
    /**
     * 初始化令牌桶 一次性放入足够的令牌
     *
     * @param $redis
     * @param $url
     * @param $limit
     * @return void
     */
    public function initTokenBucket($redis, $url, $limit)
    {
        $redis->del($url);
        $this->addToken($redis, $url, $limit);
    }
    /**
     * 获取令牌
     *
     * @param $redis
     * @param $url
     * @return bool
     */
    public function getToken($redis, $url)
    {
        return $redis->rpop($url) ? true : false;
    }
    /**
     * 根据一定时间来发放令牌，本来是想实现毫秒级的定时发送指定数量令牌
     * 想想太复杂，遂改成指定秒数发放足额令牌形式
     * 比如限定是1秒100请求，超过1秒了就补全缺的令牌使可用数达到100即可
     *
     * @param $redis
     * @param $url
     * @param $limit
     * @return boolean
     */
    public function addToken($redis, $url, $limit)
    {
        $size = (int)$redis->lLen($url);
        $fillNum = $limit - $size;
        if ($fillNum > 0) {
            $token = array_fill(0, $fillNum, 1);
            $redis->lpush($url, $token);
        }
        return true;
    }
    /**
     * 重新计算时间戳 毫秒
     *
     * @return float
     */
    public function getMicroSecond()
    {
        list($microsecond, $second) = explode(' ', microtime());
        $microsecond = (float)sprintf('%.0f', (floatval($microsecond) + floatval($second)) * 1000);
        return $microsecond;
    }
}

漏桶限流

漏桶算法大致如下：

漏桶算法稍微不同与令牌桶算法的一点是：对于取令牌的频率也有限制，要按照 t/n 固定的速度来取令牌，所以可以看出漏桶算法对流量的整形效果更加好，流量更加平滑，任何突发流量都会被限流。因为令牌桶大小为 b，所以是可以应对突发流量的。这个我就没有去实现了！有时间在补

总结

对于令牌桶的算法虽然可以很好的限制流量，但是它是否决式的，会导致误杀很多正常的请求。所以令牌桶和漏桶算法比较适合阻塞式限流，比如一些后台 job 类的限流，超过了最大访问频率之后，请求并不会被拒绝，而是会被阻塞到有令牌后再继续执行。对于像微服务接口这种对响应时间比较敏感的限流场景，会比较适合选择基于时间窗口的否决式限流算法，其中滑动时间窗口限流算法空间复杂度较高，内存占用会比较多，所以对比来看，尽管固定时间窗口算法处理临界突发流量的能力较差，但实现简单，而简单带来了好的性能和不容易出错，所以固定时间窗口算法也不失是一个好的微服务接口限流算法。

-------------The End-------------