系列文章Redis应用-分布式锁Redis应用-异步消息队列和延迟队列Redis应用-位图Redis应用-HyperLogLog高并发场景保护系统的三大利器：缓存、降级、和电流限制。缓存的目的是提高系统的访问速度，增加系统的处理能力；降级是在服务出现问题或影响核心进程性能时暂时阻塞。但在某些场景下，需要限制并发请求量，比如秒杀、抢购、发帖、评论、恶意爬虫等。限流算法常见的限流算法有：计数器、漏桶、令牌桶。顾名思义，计数器就是一个一个记录，然后判断限定时间窗口内的数量是否超过限制。函数isActionAllowed($userId,$action,$period,$maxCount){$redis=newRedis();$redis->connect('127.0.0.1',6379);$key=sprintf('hist:%s:%s',$userId,$action);$现在=毫秒时间（）；#毫秒时间戳$pipe=$redis->multi(Redis::PIPELINE);//使用管道提高性能$pipe->zadd($key,$now,$now);//value和score都使用毫秒时间戳$pipe->zremrangebyscore($key,0,$now-$period);//去掉时间窗之前的行为记录，剩下的就是$pipe->zcard($key);//获取窗口中的行为数$pipe->expire($key,$period+1);//在过期时间上加一秒$replies=$pipe->exec();返回$replies[2]<=$maxCount;}for($i=0;$i<20;$i++){var_dump(isActionAllowed("110","reply",60*1000,5));//执行可以发现只传了前5次}//返回当前毫秒时间戳functionmsectime(){list($msec,$sec)=explode('',microtime());$msectime=(float)sprintf('%.0f',(floatval($msec)+floatval($sec))*1000);返回$毫秒；}漏桶（LeakyBucket）算法思路很简单，水（请求）先进入漏桶，漏桶以一定的速度离开水（接口有响应率），当进水速度过快时，会直接溢出（访问频率超过接口响应率），然后拒绝请求。可见，漏桶算法可以强制限制数据传输速率。示意图如下：具体代码实现如下capacity=$capacity;//漏斗容量$this->leakingRate=$leakingRate;//漏斗流量$this->leftQuote=$capacity;//漏斗的剩余空间$this->leakingTs=time();//最后一次泄漏的时间}publicfunctionmakeSpace(){$now=time();$deltaTs=$now-$this->leakingTs;//自上次泄漏以来已经过了多长时间$deltaQuota=$deltaTs*$this->leakingRate;//可用空间if($deltaQuota<1){return;}$this->leftQuote+=$deltaQuota;//增加剩余空间$this->leakingTs=time();//记录漏水时间if($this->leftQuota>$this->capacaty){$this->leftQuote=$this->capacity;}}公共函数watering($quota){$this->makeSpace();//泄漏操作if($this->leftQuote>=$quota){$this->leftQuote-=$quota;返回真；}返回假；}}$funnels=[];global$funnel;functionisActionAllowed($userId,$action,$capacity,$leakingRate){$key=sprintf("%s:%s",$userId,$action);$funnel=$GLOBALS['漏斗'][$key]??'';如果(!$funnel){$funnel=newFunnel($capacity,$leakingRate);$GLOBALS['漏斗'][$key]=$漏斗；}返回$funnel->watering(1);}for($i=0;$i<20;$i++){var_dump(isActionAllowed("110","reply",15,0.5));//执行可以发现只有前15次通过了}核心逻辑是makeSpace，在每次灌水前调用，触发漏水，为漏斗腾出空间。我们可以使用Redis中的hash结构来存储对应的字段，在灌水的时候将字段取出来进行逻辑计算后，存储到hash结构中，完成一次行为频率检测。但是有个问题就是整个过程的原子性无法保证，也就是说必须要用锁来控制，但是如果锁失败了，就得重试或者放弃，会导致性能下降，影响用户经验，代码复杂度也会增加。Elevated，此时Redis提供了一个插件，出现了Redis-Cell。Redis-CellRedis4.0提供了一个Redis限流模块redis-cell，它提供了漏斗算法和原子限流指令。该模块只有一条指令cl.throttle，其参数和返回值比较复杂。>cl.throttletom:reply14306011)(integer)0#0表示允许，1表示拒绝2)(integer)15#料斗容量capacity3)(integer)14#料斗剩余空间left_quota4)(integer)-1#如果被拒绝，需要多长时间重试，单位秒5)(integer)2#漏斗完全清空需要多长时间，单位秒这个命令的意思是允许用户tom的回复行为的频率是每60s最多30次，漏斗的初始容量为15（因为它是从0开始计数的，从15到14），每个行为默认占用的空间为1（可选参数）。如果被拒绝，则取返回数组的第4个值作为重试时间执行sleep，也可以用异步定时任务重试。令牌桶令牌桶算法（TokenBucket）与LeakyBucket作用相同但方向相反，更容易理解。随着时间的推移，系统会按照恒定的1/QPS时间间隔（如果QPS=100，则间隔为10ms）向桶中添加Token（想象漏水的反面，水龙头在不断加水），如果桶已满，不再添加。当有新的请求到来时，每一个都会拿走一个Token，如果没有Token可以拿走令牌桶的另一个好处是可以方便的改变速度。一旦需要提速，则根据需要提速放入桶中的代币。一般会每隔一段时间（比如100毫秒）发送到桶中。添加一定数量的代币，一些变体算法会实时计算应该添加的代币数量。具体实现参考php基于redis使用令牌桶算法实现流控_config=$config;$this->_queue=$队列；$this->_max=$max;$this->_redis=$this->connect();}/***jointoken*@paramInt$num添加的token个数*@returnIntjoin个数*/publicfunctionadd($num=0){//当前剩余的token个数$curnum=intval($this->_redis->lSize($this->_queue));//最大限度令牌数量$maxnum=intval($this->_max);//计算最多可以添加的token个数，不能超过最大token个数$num=$maxnum>=$curnum+$num?$num:$maxnum-$curnum;//添加标记if($num>0){$token=array_fill(0,$num,1);$this->_redis->lPush($this->_queue,...$token);返回$num;}返回0;}/***获取令牌*@returnBoolean*/publicfunctionget(){return$this->_redis->rPop($this->_queue)?真假;}/***重置令牌桶并用令牌填充它*/publicfunctionreset(){$this->_redis->delete($this->_queue);$this->add($this->_max);}privatefunctionconnect(){try{$redis=newRedis();$redis->connect($this->_config['host'],$this->_config['port'],$this->_config['timeout'],$this->_config['reserved'],$this->_config['retry_interval']);如果（空（$this->_config['auth'])){$redis->auth($this->_config['auth']);$redis->select($this->_config['index']);}catch(\RedisException$e){thrownewException($e->getMessage());返回假；}返回$redis；}}$config=array('host'=>'localhost','port'=>6379,'index'=>0,'auth'=>'','timeout'=>1,'reserved'=>NULL,'retry_interval'=>100,);//令牌桶容器$queue='mycontainer';//最大令牌数$max=5;//创建一个TrafficShaper对象$oTrafficShaper=newTrafficShaper($config,$queue,$max);//重置令牌桶并填充令牌$oTrafficShaper->reset();//循环获取令牌，令牌桶中只有5个令牌，所以最后3次获取失败for($i=0;$i<8;$i++){var_dump($oTrafficShaper->get()));}//添加10个token，最大token为5，所以只能添加5$add_num=$oTrafficShaper->add(10);var_dump($add_num);//循环获取token，只有令牌桶中有5个令牌，所以最后一次获取失败for($i=0;$i<6;$i++){var_dump($oTrafficShaper->get());}?>本文同时发布在微信公众号【小刀资讯】，欢迎扫描二维码关注！
