Redis應用-限流

系列文章

• Redis應用-分佈式鎖
• Redis應用-異步消息隊列與延時隊列
• Redis應用-位圖
• Redis應用-HyperLogLog
• Redis應用-布隆過濾器
• Redis應用-限流
• Redis應用-Geo

在高併發場景下有三把利器保護系統：緩存、降級、和限流。緩存的目的是提高系統的訪問你速度和增大系統能處理的容量；降級是當服務出問題或影響到核心流程的性能則須要暫時屏蔽掉。而有些場景則須要限制併發請求量，如秒殺、搶購、發帖、評論、惡意爬蟲等。

限流算法
常見的限流算法有：計數器，漏桶、令牌桶。

計數器

顧名思義就是來一個記一個，而後判斷在有限時間窗口內的數量是否超過限制便可

function isActionAllowed($userId, $action, $period, $maxCount) 
{
    $redis = new Redis();
    $redis->connect('127.0.0.1', 6379);
    $key = sprintf('hist:%s:%s', $userId, $action);
    $now = msectime();   # 毫秒時間戳

    $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();
    return $replies[2] <= $maxCount;
}
for ($i=0; $i<20; $i++){
    var_dump(isActionAllowed("110", "reply", 60*1000, 5)); //執行能夠發現只有前5次是經過的
}

//返回當前的毫秒時間戳
function msectime() {
    list($msec, $sec) = explode(' ', microtime());
    $msectime = (float)sprintf('%.0f', (floatval($msec) + floatval($sec)) * 1000);
    return $msectime;
 }

漏桶

漏桶(Leaky Bucket)算法思路很簡單,水(請求)先進入到漏桶裏,漏桶以必定的速度出水(接口有響應速率),當水流入速度過大會直接溢出(訪問頻率超過接口響應速率),而後就拒絕請求,能夠看出漏桶算法能強行限制數據的傳輸速率.示意圖以下:

具體代碼實現以下

<?php

class Funnel {

    private $capacity;
    private $leakingRate;
    private $leftQuote;
    private $leakingTs;

    public function __construct($capacity, $leakingRate)
    {
        $this->capacity = $capacity;    //漏斗容量
        $this->leakingRate = $leakingRate;//漏斗流水速率
        $this->leftQuote = $capacity; //漏斗剩餘空間
        $this->leakingTs = time(); //上一次漏水時間
    }

    public function makeSpace()
    {
        $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;
        }
    }

    public function watering($quota)
    {
        $this->makeSpace(); //漏水操做
        if($this->leftQuote >= $quota) {
            $this->leftQuote -= $quota;
            return true;
        }
        return false;
    }
}


$funnels = [];
global $funnel;

function isActionAllowed($userId, $action, $capacity, $leakingRate)
{
    $key = sprintf("%s:%s", $userId, $action);
    $funnel = $GLOBALS['funnel'][$key] ?? '';
    if (!$funnel) {
        $funnel  = new Funnel($capacity, $leakingRate);
        $GLOBALS['funnel'][$key] = $funnel;
    }
    return $funnel->watering(1);
}

for ($i=0; $i<20; $i++){
    var_dump(isActionAllowed("110", "reply", 15, 0.5)); //執行能夠發現只有前15次是經過的
}

核心邏輯就是makeSpace，在每次灌水前調用以觸發漏水，給漏斗騰出空間。
funnels咱們能夠利用Redis中的hash結構來存儲對應字段，灌水時將字段取出進行邏輯運算後再存入hash結構中便可完成一次行爲頻度的檢測。但這有個問題就是整個過程的原子性沒法保證，意味着要用鎖來控制，但若是加鎖失敗，就要重試或者放棄，這回致使性能降低和影響用戶體驗，同時代碼複雜度也升高了，此時Redis提供了一個插件，Redis-Cell出現了。

Redis-Cell

Redis 4.0提供了一個限流Redis模塊，名稱爲redis-cell，該模塊提供漏斗算法，並提供原子的限流指令。

該模塊只有一條指令cl.throttle，其參數和返回值比較複雜。

> cl.throttle tom:reply 14 30 60 1
1) (integer) 0    # 0表示容許，1表示拒絕
2) (integer) 15    # 漏斗容量capacity
3) (integer) 14    # 漏斗剩餘空間left_quota
4) (integer) -1    # 若是拒絕了，須要多長時間後再重試，單位秒
5) (integer) 2    # 多長時間後，漏斗徹底空出來，單位秒

該指令意思爲，容許用戶tom的reply行爲的頻率爲每60s最多30次，漏斗初始容量爲15（由於是從0開始計數，到14爲15個），默認每一個行爲佔據的空間爲1（可選參數）。
若是被拒絕，取返回數組的第四個值進行sleep便可做爲重試時間，也能夠異步定時任務來重試。

令牌桶

令牌桶算法(Token Bucket)和 Leaky Bucket 效果同樣但方向相反的算法,更加容易理解.隨着時間流逝,系統會按恆定1/QPS時間間隔(若是QPS=100,則間隔是10ms)往桶裏加入Token(想象和漏洞漏水相反,有個水龍頭在不斷的加水),若是桶已經滿了就再也不加了.新請求來臨時,會各自拿走一個Token,若是沒有Token可拿了就阻塞或者拒絕服務.

令牌桶的另一個好處是能夠方便的改變速度. 一旦須要提升速率,則按需提升放入桶中的令牌的速率. 通常會定時(好比100毫秒)往桶中增長必定數量的令牌, 有些變種算法則實時的計算應該增長的令牌的數量.

具體實現可參考php 基於redis使用令牌桶算法實現流量控制

<?php
class TrafficShaper
{ 
    private $_config; // redis設定
    private $_redis;  // redis對象
    private $_queue;  // 令牌桶
    private $_max;    // 最大令牌數

    /**
     * 初始化
     * @param Array $config redis鏈接設定
     */
    public function __construct($config, $queue, $max)
    {
        $this->_config = $config;
        $this->_queue = $queue;
        $this->_max = $max;
        $this->_redis = $this->connect();
    }

    /**
     * 加入令牌
     * @param Int $num 加入的令牌數量
     * @return Int 加入的數量
     */
    public function add($num = 0)
    {
        // 當前剩餘令牌數
        $curnum = intval($this->_redis->lSize($this->_queue));
        // 最大令牌數
        $maxnum = intval($this->_max);
        // 計算最大可加入的令牌數量，不能超過最大令牌數
        $num = $maxnum >= $curnum + $num ? $num : $maxnum - $curnum;
        // 加入令牌
        if ($num > 0) {
            $token = array_fill(0, $num, 1);
            $this->_redis->lPush($this->_queue, ...$token);
            return $num;
        }
        return 0;
    }

    /**
     * 獲取令牌
     * @return Boolean
     */
    public function get()
    {
        return $this->_redis->rPop($this->_queue) ? true : false;
    }

    /**
     * 重設令牌桶，填滿令牌
     */
    public function reset()
    {
        $this->_redis->delete($this->_queue);
        $this->add($this->_max);
    }

    private function connect()
    {
        try {
            $redis = new Redis();
            $redis->connect($this->_config['host'], $this->_config['port'], $this->_config['timeout'], $this->_config['reserved'], $this->_config['retry_interval']);
            if (empty($this->_config['auth'])) {
                $redis->auth($this->_config['auth']);
            }
            $redis->select($this->_config['index']);
        } catch (\RedisException $e) {
            throw new Exception($e->getMessage());
            return false;
        }
        return $redis;
    }
} 

$config = array(
    'host' => 'localhost',
    'port' => 6379,
    'index' => 0,
    'auth' => '',
    'timeout' => 1,
    'reserved' => NULL,
    'retry_interval' => 100,
);
// 令牌桶容器
$queue = 'mycontainer';
 // 最大令牌數
$max = 5;
// 建立TrafficShaper對象
$oTrafficShaper = new TrafficShaper($config, $queue, $max);
// 重設令牌桶，填滿令牌
$oTrafficShaper->reset();
// 循環獲取令牌，令牌桶內只有5個令牌，所以最後3次獲取失敗
for ($i = 0; $i < 8; $i++) {
    var_dump($oTrafficShaper->get());
}
// 加入10個令牌，最大令牌爲5，所以只能加入5個
$add_num = $oTrafficShaper->add(10);
var_dump($add_num);
// 循環獲取令牌，令牌桶內只有5個令牌，所以最後1次獲取失敗
for ($i = 0; $i < 6; $i++) {
    var_dump($oTrafficShaper->get());
}
?>

本文亦在微信公衆號【小道資訊】發佈，歡迎掃碼關注！