Приложение Redis-Ограничение по Току

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 ();  millisecond timestamp

    $pipe = $redis - > multi (Redis:: PIPELINE); // Pipeline for performance improvement
    $pipe - > zadd ($key, $now, $now); both // value and score use millisecond timestamps
    $pipe - > zremrange by score ($key, 0, $now - $period); // Remove the behavior record before the time window, and the rest is in the time window
    $pipe - > zcard ($key); // Get the number of actions in the window
    $pipe - > expire ($key, $period + 1); // plus one second expiration time
    $replies = $pipe->exec();
    return $replies[2] <= $maxCount;
}
for ($i=0; $i<20; $i++){
    Var_dump (isAction Allowed ("110", "reply", 60*1000, 5)); // Execution can find that only the first five passes.
}

// Returns the current millisecond timestamp
function msectime() {
    list($msec, $sec) = explode(' ', microtime());
    $msectime = (float)sprintf('%.0f', (floatval($msec) + floatval($sec)) * 1000);
    return $msectime;
 }

php

class Funnel {

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

    public function __construct($capacity, $leakingRate)
    {
        This - > capacity = capacity; // funnel capacity
        This - > leaking Rate = leaking Rate; // funnel flow rate
        $this - > leftQuote = $capacity; // funnel remaining space
        This - > leaking Ts = time (); // last leak time
    }

    public function makeSpace()
    {
        $now = time();
        $deltaTs = $now - $this - > leaking Ts; // How long has it been since the last leak?
        $deltaQuota = $deltaTs * $this - > leaking Rate; // freeable space
        if($deltaQuota < 1) {  
            return;
        }
        $this - > leftQuote += $deltaQuota; // Increase the remaining space
        $this - > leaking Ts = time (); // record leak time
        if($this->leftQuota > $this->capacaty){
            $this->leftQuote = $this->capacity;
        }
    }

    public function watering($quota)
    {
        This - > makeSpace (); // Leakage operation
        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 (isAction Allowed ("110", "reply", 15, 0.5)); // Execution can find that only the first 15 passes
}

> cl.throttle tom:reply 14 30 60 1
1) Integer 0# 0 denotes permission and 1 denies refusal.
2) Integer 15 # funnel capacity
3) Integer 14 # funnel remaining space left_quota
4) Integer - 1 # If rejected, how long will it take to retry in unit seconds?
5) How long after integer 2, the funnel is completely empty, in unit seconds

_config = $config;
        $this->_queue = $queue;
        $this->_max = $max;
        $this->_redis = $this->connect();
    }

    /**
     * Added token
     *@ Number of tokens added by param Int $num
     *@ Number of return Int additions
     */
    public function add($num = 0)
    {
        // Current number of remaining tokens
        $curnum = intval($this->_redis->lSize($this->_queue));
        // Maximum number of tokens
        $maxnum = intval($this->_max);
        // Calculate the maximum number of tokens that can be added, not exceeding the maximum number of tokens
        $num = $maxnum >= $curnum + $num ? $num : $maxnum - $curnum;
        // Added token
        if ($num > 0) {
            $token = array_fill(0, $num, 1);
            $this->_redis->lPush($this->_queue, ...$token);
            return $num;
        }
        return 0;
    }

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

    /**
     * Reset the token bucket and fill it with tokens
     */
    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,
);
// Token Bucket Container
$queue = 'mycontainer';
 // Maximum number of tokens
$max = 5;
// Create TrafficShaper objects
$oTrafficShaper = new TrafficShaper($config, $queue, $max);
// Reset the token bucket and fill it with tokens
$oTrafficShaper->reset();
// The last three acquisitions fail because there are only five tokens in the token bucket.
for ($i = 0; $i < 8; $i++) {
    var_dump($oTrafficShaper->get());
}
// Add 10 tokens, the maximum token is 5, so only 5 tokens can be added.
$add_num = $oTrafficShaper->add(10);
var_dump($add_num);
// The last acquisition failed because there were only five tokens in the token bucket.
for ($i = 0; $i < 6; $i++) {
    var_dump($oTrafficShaper->get());
}
?>