限流组件设计与实现

PHP

计数器算法

原理：如果服务的QPS为1000，那么在1s的时间范围内，每次请求到来计数器加1，超过1000则拒绝请求。

实现：

<?php
/**
 * Creator: donovanliu
 */
class Counter
{
    private static $rateLimit = 3000;
    public static function getRedisName($ip) {
        $redisNameMap = array(
            'doki',
            'user_property'
        );
        $num = crc32($ip);
        $num = abs($num % 2);
        return $redisNameMap[$num];
    }
    public static function pass($num) {
        $ip = $_SERVER['SERVER_ADDR'];
        $redis = TMPHPRedisClientFactory::getClient(self::getRedisName($ip));
        $key = 'rate_limit_' . $ip . '_' . time();
        if ($redis->incrBy($key, $num) > self::$rateLimit) {
            $redis->decrBy($key, $num);
            return false;
        }
        return true;
    }
}

缺点：如果在1s内的前100ms，已经通过了1000个请求，那后面的900ms将无法通过任何请求，从而产生突刺现象。

漏斗桶算法

原理：漏斗桶算法可以通过限制平均速率消除突刺现象。每个请求到来时，就向水滴一样加入漏斗桶，满了就溢出，也就是拒绝请求。漏斗桶以固定的速率漏出水滴，通过请求。

实现：

<?php
/**
 * Creator: donovanliu
 */
class FunnelBucket
{
    private static $rateLimit = 3000;
    public static function getRedisName($ip) {
        $redisNameMap = array(
            'doki',
            'user_property'
        );
        $num = crc32($ip);
        $num = abs($num % 2);
        return $redisNameMap[$num];
    }
    public static function pass($num) {
        $ip = $_SERVER['SERVER_ADDR'];
        $redis = TMPHPRedisClientFactory::getClient(self::getRedisName($ip));
        $key = 'rate_limit_' . $ip;
        if ($redis->incrBy($key, $num) > self::$rateLimit) {
            $redis->decrBy($key, $num);
            return false;
        }
        $lock = 'rate_limit_' . $ip . '_lock';
        while (!$redis->setNX($lock, 'whatever')) {
            usleep(10);
        }
        $redis->pexpire($key, $num);
        $redis->decrBy($key, $num);
        return true;
    }
}

缺点：无法应对短时间内的突发流量。

令牌桶算法

原理：令牌桶算法可以在限制平均速率的同时允许一定程度的并发，从而应对突发流量。令牌桶算法以固定的速率向桶里放入令牌，超过上限就丢弃。请求到来时，需要先获取令牌，拿到令牌就可以通过，否则就拒绝请求。

实现：

<?php
/**
 * Creator: donovanliu
 */
class TokenBucket
{
    private static $ipArr = array(
    );
    private static $rateLimit = 3000;
    private static $timeInterval = 100;
    public static function getRedisName($ip) {
        $redisNameMap = array(
            'doki',
            'user_property'
        );
        $num = crc32($ip);
        $num = abs($num % 2);
        return $redisNameMap[$num];
    }
    public static function getToken($num) {
        $runStartTime = microtime_float();
        $ip = $_SERVER['SERVER_ADDR'];
        $redis = TMPHPRedisClientFactory::getClient(self::getRedisName($ip));
        $key = 'rate_limit_' . $ip;
        if ($redis->decrBy($key, $num) < 0) {
            $redis->incrBy($key, $num);
            $runEndTime = microtime_float();
            modstr_passive_report("ratelimiter", "chicken_$num", 708, ($runEndTime-$runStartTime)*1000000, -1, '');
            return false;
        }
        $runEndTime = microtime_float();
        modstr_passive_report("ratelimiter", "chicken_$num", 708, ($runEndTime-$runStartTime)*1000000, 0, '');
        return true;
    }
    public static function check() {
        $ipArr = self::$ipArr;
        foreach ($ipArr as $ip) {
            $redis = TMPHPRedisClientFactory::getClient(self::getRedisName($ip));
            $key = 'rate_limit_' . $ip;
            echo $redis->get($key) . "\n";
        }
    }
    public static function init() {
        $ipArr = self::$ipArr;
        $rateLimit = self::$rateLimit;
        if (!is_dir("/data/logs/ratelimiter/")) {
            mkdir("/data/logs/ratelimiter/", 0777);
        }
        $log =  "/data/logs/ratelimiter/producer_init.log";
        foreach ($ipArr as $ip) {
            $redis = TMPHPRedisClientFactory::getClient(self::getRedisName($ip));
            $key = 'rate_limit_' . $ip;
            $redis->set($key, $rateLimit);
            error_log(date('Y-m-d H:i:s') . " | produce token | $rateLimit\r\n", 3, $log);
        }
        error_log(date('Y-m-d H:i:s') . " | produce token end\r\n", 3, $log);
    }
    public static function produce() {
        $ipArr = self::$ipArr;
        $rateLimit = self::$rateLimit;
        $timeInterval = self::$timeInterval;
        $addInterval = $rateLimit / $timeInterval;
        $date = date( 'Ymd_H' );
        $log =  "/data/logs/ratelimiter/producer_$date.log";
        foreach ($ipArr as $ip) {
            $redis = TMPHPRedisClientFactory::getClient(self::getRedisName($ip));
            $key = 'rate_limit_' . $ip;
            $rateRest = $redis->get($key);
            $canAdd = $rateLimit - $rateRest;
            if ($canAdd < $addInterval) {
                $redis->incrBy($key, $canAdd);
                if (!empty($canAdd)) {
                    error_log(date('Y-m-d H:i:s') . " | produce token | $canAdd\r\n", 3, $log);
                }
            } else {
                $redis->incrBy($key, $addInterval);
                error_log(date('Y-m-d H:i:s') . " | produce token | $addInterval\r\n", 3, $log);
            }
        }
    }
}

<?php
require_once(ROOT_PATH . "/components/ratelimiter/tokenBucket.php");
try{
    TMConfig::initialize();
    //TokenBucket::check();die;
    TokenBucket::init();
    swoole_timer_tick(100, function ($timer_id) {
        //echo "tick-100ms\n";
        TokenBucket::produce();
    });
}catch(TMException $e){
}