PHP `Rate Limiting` (限流) 算法 (`令牌桶`/`漏桶`) 在高并发 API 中的实现

各位观众老爷们，晚上好！我是你们的老朋友，BUG终结者，今天要给大家带来一场关于PHP在高并发API中如何玩转“限流”的盛宴。这次咱们不来虚的，直接上干货，手把手教你用“令牌桶”和“漏桶”算法，让你的API在高并发的浪潮中稳如老狗！

开场白：为啥要限流？

首先，咱们得搞清楚，为什么要限流？想象一下，你的API就像一个水龙头，用户请求就像水。如果水龙头一直开着，水管可能爆掉，你的服务器可能瘫痪。限流就是给水龙头加个阀门，控制水的流量，保证水管（服务器）的安全。

在高并发场景下，没有限流的API就像一个没穿裤子的小伙子，很容易被人扒个精光！

第一部分：令牌桶算法 (Token Bucket)

令牌桶算法，顾名思义，就是有一个装满令牌的桶。每个请求过来，都要从桶里拿一个令牌。如果桶里没令牌了，那就拒绝请求。

核心思想： 以恒定速率向桶中放入令牌，请求到来时尝试从桶中获取令牌，获取成功则放行，否则丢弃或排队等待。
优点： 允许一定程度的突发流量，因为桶里可以积攒一些令牌。
缺点： 实现相对复杂。

代码实现：

咱们先来个最简单的内存版令牌桶：

<?php

class TokenBucket
{
    private $capacity;  // 桶的容量
    private $rate;      // 令牌生成速率 (令牌/秒)
    private $tokens;    // 当前令牌数量
    private $lastRefillTimestamp; // 上次填充令牌的时间戳

    public function __construct(int $capacity, float $rate)
    {
        $this->capacity = $capacity;
        $this->rate = $rate;
        $this->tokens = $capacity; // 初始时桶是满的
        $this->lastRefillTimestamp = microtime(true);
    }

    /**
     * 尝试获取一个令牌
     *
     * @return bool true: 获取成功, false: 获取失败
     */
    public function tryConsume(int $tokensToConsume = 1): bool
    {
        $this->refill(); // 先填充令牌

        if ($this->tokens >= $tokensToConsume) {
            $this->tokens -= $tokensToConsume;
            return true;
        }

        return false;
    }

    /**
     * 填充令牌
     */
    private function refill(): void
    {
        $now = microtime(true);
        $elapsedTime = $now - $this->lastRefillTimestamp;
        $newTokens = $elapsedTime * $this->rate;

        // 避免浮点数精度问题
        $this->tokens = min($this->capacity, $this->tokens + $newTokens);
        $this->lastRefillTimestamp = $now;
    }

    /**
     * 获取当前令牌数量
     *
     * @return float
     */
    public function getTokens(): float
    {
        $this->refill();
        return $this->tokens;
    }
}

// 使用示例：
$bucket = new TokenBucket(10, 2); // 容量为10，每秒生成2个令牌

for ($i = 0; $i < 15; $i++) {
    if ($bucket->tryConsume()) {
        echo "Request {$i}: Passed! Tokens left: " . $bucket->getTokens() . "n";
    } else {
        echo "Request {$i}: Rejected!n";
    }
    usleep(200000); // 模拟请求间隔 (0.2秒)
}

?>

代码解释：

capacity: 桶的容量，决定了最多能积攒多少令牌，也决定了允许的最大突发流量。
rate: 令牌生成速率，决定了平均允许的请求速率。
tokens: 当前桶里有多少令牌。
lastRefillTimestamp: 上次填充令牌的时间戳，用于计算应该填充多少令牌。
tryConsume(): 尝试获取令牌，如果桶里有足够的令牌，就减掉相应的数量，并返回true。否则，返回false。
refill(): 填充令牌，根据时间和速率计算应该填充多少令牌，并更新桶里的令牌数量。

进阶版：Redis令牌桶

上面的代码只能在单进程中使用，在高并发场景下，我们需要一个共享的令牌桶，Redis就派上用场了。

<?php

class RedisTokenBucket
{
    private $redis;
    private $bucketKey;
    private $capacity;
    private $rate;

    public function __construct(Redis $redis, string $bucketKey, int $capacity, float $rate)
    {
        $this->redis = $redis;
        $this->bucketKey = $bucketKey;
        $this->capacity = $capacity;
        $this->rate = $rate;
    }

    /**
     * 尝试获取一个令牌
     *
     * @return bool true: 获取成功, false: 获取失败
     */
    public function tryConsume(int $tokensToConsume = 1): bool
    {
        $now = microtime(true);
        $this->refill($now);

        $tokens = $this->redis->get($this->bucketKey);
        if ($tokens === false) {
            $tokens = $this->capacity; // 初始令牌数量
        }
        $tokens = (float)$tokens;

        if ($tokens >= $tokensToConsume) {
            $newTokens = $tokens - $tokensToConsume;
            $this->redis->set($this->bucketKey, $newTokens);
            return true;
        }

        return false;
    }

    /**
     * 填充令牌
     */
    private function refill(float $now): void
    {
        $lastRefillTimestamp = $this->redis->get("{$this->bucketKey}:last_refill");
        if ($lastRefillTimestamp === false) {
            $lastRefillTimestamp = $now; // 首次填充
        }

        $lastRefillTimestamp = (float)$lastRefillTimestamp;
        $elapsedTime = $now - $lastRefillTimestamp;
        $newTokens = $elapsedTime * $this->rate;

        if ($newTokens > 0) {
            $tokens = $this->redis->get($this->bucketKey);
            if ($tokens === false) {
                $tokens = $this->capacity; // 初始令牌数量
            }
            $tokens = (float)$tokens;

            $this->redis->set($this->bucketKey, min($this->capacity, $tokens + $newTokens));
            $this->redis->set("{$this->bucketKey}:last_refill", $now);
        }
    }
}

// 使用示例：
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

$bucket = new RedisTokenBucket($redis, 'my_api:token_bucket', 10, 2); // 容量为10，每秒生成2个令牌

for ($i = 0; $i < 15; $i++) {
    if ($bucket->tryConsume()) {
        echo "Request {$i}: Passed!n";
    } else {
        echo "Request {$i}: Rejected!n";
    }
    usleep(200000); // 模拟请求间隔 (0.2秒)
}

?>

代码解释：

redis: Redis连接实例。
bucketKey: 用于存储令牌数量的Redis key。
{$this->bucketKey}:last_refill: 用于存储上次填充令牌的时间戳的Redis key。
其他参数和方法与内存版令牌桶类似，只是将令牌数量和时间戳存储在Redis中。

注意事项：

Redis操作需要考虑网络延迟和Redis本身的性能。
可以使用Redis的事务或Lua脚本来保证get和set操作的原子性，避免并发问题。
可以根据实际情况调整桶的容量和令牌生成速率。

第二部分：漏桶算法 (Leaky Bucket)

漏桶算法就像一个底部有小孔的桶。请求就像倒入桶里的水，桶以恒定的速率漏水。如果水倒入的速度太快，桶会溢出，溢出的水就代表被拒绝的请求。

核心思想： 请求先进入桶中，然后以恒定速率从桶中流出，从而平滑突发流量。
优点： 实现简单，可以严格限制请求的速率。
缺点： 无法应对突发流量，所有请求都必须等待。

代码实现：

同样，先来个简单的内存版漏桶：

<?php

class LeakyBucket
{
    private $capacity;  // 桶的容量
    private $leakRate;  // 漏水速率 (请求/秒)
    private $queue = []; // 请求队列
    private $lastLeakTimestamp; // 上次漏水的时间戳

    public function __construct(int $capacity, float $leakRate)
    {
        $this->capacity = $capacity;
        $this->leakRate = $leakRate;
        $this->lastLeakTimestamp = microtime(true);
    }

    /**
     * 尝试添加一个请求到桶中
     *
     * @return bool true: 添加成功, false: 添加失败
     */
    public function tryAdd(): bool
    {
        $this->leak(); // 先漏水

        if (count($this->queue) < $this->capacity) {
            $this->queue[] = microtime(true); // 将当前时间戳添加到队列
            return true;
        }

        return false;
    }

    /**
     * 漏水
     */
    private function leak(): void
    {
        $now = microtime(true);
        $elapsedTime = $now - $this->lastLeakTimestamp;
        $leakedRequests = $elapsedTime * $this->leakRate;

        // 避免浮点数精度问题
        $leakedRequests = (int)$leakedRequests;

        for ($i = 0; $i < $leakedRequests; $i++) {
            if (!empty($this->queue)) {
                array_shift($this->queue); // 移除队列头部的请求
            } else {
                break;
            }
        }

        $this->lastLeakTimestamp = $now;
    }
}

// 使用示例：
$bucket = new LeakyBucket(5, 1); // 容量为5，每秒漏1个请求

for ($i = 0; $i < 10; $i++) {
    if ($bucket->tryAdd()) {
        echo "Request {$i}: Accepted! Queue size: " . count($bucket->queue) . "n";
    } else {
        echo "Request {$i}: Rejected!n";
    }
    usleep(500000); // 模拟请求间隔 (0.5秒)
}

?>

代码解释：

capacity: 桶的容量，决定了最多能容纳多少请求。
leakRate: 漏水速率，决定了每秒处理多少请求。
queue: 请求队列，存储等待处理的请求。
lastLeakTimestamp: 上次漏水的时间戳，用于计算应该漏掉多少请求。
tryAdd(): 尝试添加请求到桶中，如果桶未满，则将请求添加到队列中，并返回true。否则，返回false。
leak(): 漏水，根据时间和速率计算应该漏掉多少请求，并从队列头部移除相应的请求。

进阶版：Redis漏桶

同样，为了在高并发场景下使用，我们需要一个共享的漏桶，Redis再次登场。

<?php

class RedisLeakyBucket
{
    private $redis;
    private $bucketKey;
    private $capacity;
    private $leakRate;

    public function __construct(Redis $redis, string $bucketKey, int $capacity, float $leakRate)
    {
        $this->redis = $redis;
        $this->bucketKey = $bucketKey;
        $this->capacity = $capacity;
        $this->leakRate = $leakRate;
    }

    /**
     * 尝试添加一个请求到桶中
     *
     * @return bool true: 添加成功, false: 添加失败
     */
    public function tryAdd(): bool
    {
        $now = microtime(true);
        $this->leak($now);

        $queueSize = $this->redis->llen($this->bucketKey);
        if ($queueSize === false) {
            $queueSize = 0;
        }

        if ($queueSize < $this->capacity) {
            $this->redis->rpush($this->bucketKey, $now); // 将当前时间戳添加到队列尾部
            return true;
        }

        return false;
    }

    /**
     * 漏水
     */
    private function leak(float $now): void
    {
        $lastLeakTimestamp = $this->redis->get("{$this->bucketKey}:last_leak");
        if ($lastLeakTimestamp === false) {
            $lastLeakTimestamp = $now; // 首次漏水
        }

        $lastLeakTimestamp = (float)$lastLeakTimestamp;
        $elapsedTime = $now - $lastLeakTimestamp;
        $leakedRequests = $elapsedTime * $this->leakRate;

        // 避免浮点数精度问题
        $leakedRequests = (int)$leakedRequests;

        for ($i = 0; $i < $leakedRequests; $i++) {
            // 使用 LPOP 移除队列头部的元素
            if ($this->redis->lLen($this->bucketKey) > 0) {
                $this->redis->lPop($this->bucketKey);
            }else{
                break;
            }
        }

        $this->redis->set("{$this->bucketKey}:last_leak", $now);
    }
}

// 使用示例：
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

$bucket = new RedisLeakyBucket($redis, 'my_api:leaky_bucket', 5, 1); // 容量为5，每秒漏1个请求

for ($i = 0; $i < 10; $i++) {
    if ($bucket->tryAdd()) {
        echo "Request {$i}: Accepted!n";
    } else {
        echo "Request {$i}: Rejected!n";
    }
    usleep(500000); // 模拟请求间隔 (0.5秒)
}

?>

代码解释：

redis: Redis连接实例。
bucketKey: 用于存储请求队列的Redis key（使用List数据结构）。
{$this->bucketKey}:last_leak: 用于存储上次漏水的时间戳的Redis key。
其他参数和方法与内存版漏桶类似，只是将请求队列和时间戳存储在Redis中。这里使用Redis 的List 数据结构模拟队列。 rpush 从队尾push lPop 从队头 pop

注意事项：

同样需要考虑Redis操作的性能和原子性。
可以使用Redis的事务或Lua脚本来保证操作的原子性。
漏桶算法可能会导致所有请求都需要等待，影响用户体验。

第三部分：令牌桶 vs 漏桶，谁更胜一筹？

特性	令牌桶	漏桶
突发流量	允许一定程度的突发流量	严格限制突发流量
实现难度	相对复杂	相对简单
用户体验	相对较好，允许一定程度的突发请求	可能会导致所有请求都需要等待，影响用户体验
适用场景	需要应对突发流量，但不希望超过平均速率的场景	严格限制请求速率的场景

总结：

如果你的API需要应对突发流量，但又不想超过平均速率，那么令牌桶是更好的选择。
如果你的API需要严格限制请求速率，那么漏桶是更好的选择。

第四部分：一些额外的建议

选择合适的限流算法： 根据你的业务场景和需求选择合适的限流算法。
配置合理的参数： 根据你的服务器性能和用户访问模式配置合理的桶容量和速率。
监控和告警： 监控限流效果，及时调整参数，并设置告警，以便在限流生效时及时发现问题。
友好提示： 当请求被限流时，返回友好的错误提示信息，告诉用户稍后再试。
多层限流： 可以在不同的层次进行限流，例如：
- 客户端限流： 限制单个客户端的请求速率。
- 服务端限流： 限制整个API的请求速率。
- 数据库限流： 限制数据库的查询速率。
使用中间件 很多框架都提供了限流中间件，开箱即用。例如 Laravel 的 throttle 中间件

结束语：

好了，今天的限流盛宴就到这里了。希望大家能够掌握令牌桶和漏桶算法，并在实际项目中灵活运用。记住，限流不是目的，而是手段，目的是为了保护你的API，让它在高并发的浪潮中屹立不倒！

如果你觉得这篇文章对你有帮助，请点个赞，鼓励一下我这个BUG终结者！如果你还有什么问题，欢迎在评论区留言，我会尽力解答。

下次再见！

发表回复 取消回复

发表回复取消回复