IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

PHP 令牌桶限流实现

wen PHP项目 3

本文目录导读：

PHP 令牌桶限流实现

  1. 基础令牌桶实现
  2. 高级版本（支持阻塞等待）
  3. Redis分布式版本
  4. 装饰器模式集成到业务代码
  5. 使用示例
  6. 配置文件
  7. 测试代码

我来为你实现一个完整的PHP令牌桶限流算法,包含多种实现方式和应用示例。

基础令牌桶实现

<?php
/**
 * 令牌桶限流器（单机版）
 */
class TokenBucket
{
    private $capacity;      // 桶容量（最大令牌数）
    private $rate;          // 令牌生成速率（每秒）
    private $tokens;        // 当前令牌数
    private $lastRefillTime; // 上次补充令牌的时间戳
    /**
     * @param int $capacity 桶容量
     * @param float $rate   每秒生成的令牌数
     */
    public function __construct(int $capacity, float $rate)
    {
        $this->capacity = $capacity;
        $this->rate = $rate;
        $this->tokens = $capacity; // 初始时桶是满的
        $this->lastRefillTime = microtime(true);
    }
    /**
     * 尝试获取令牌
     * @param int $requested 需要的令牌数
     * @return bool 是否成功获取
     */
    public function acquire(int $requested = 1): bool
    {
        $this->refill();
        if ($this->tokens >= $requested) {
            $this->tokens -= $requested;
            return true;
        }
        return false;
    }
    /**
     * 补充令牌
     */
    private function refill(): void
    {
        $now = microtime(true);
        $elapsed = $now - $this->lastRefillTime;
        // 计算应该补充的令牌数
        $newTokens = $elapsed * $this->rate;
        if ($newTokens > 0) {
            $this->tokens = min($this->capacity, $this->tokens + $newTokens);
            $this->lastRefillTime = $now;
        }
    }
    /**
     * 获取当前令牌数
     */
    public function getTokenCount(): float
    {
        $this->refill();
        return $this->tokens;
    }
    /**
     * 重置令牌桶
     */
    public function reset(): void
    {
        $this->tokens = $this->capacity;
        $this->lastRefillTime = microtime(true);
    }
}

高级版本（支持阻塞等待）

<?php
/**
 * 高级令牌桶（支持阻塞等待和并发控制）
 */
class AdvancedTokenBucket
{
    private $capacity;
    private $rate;
    private $tokens;
    private $lastRefillTime;
    private $mutex;  // 互斥锁
    public function __construct(int $capacity, float $rate)
    {
        $this->capacity = $capacity;
        $this->rate = $rate;
        $this->tokens = $capacity;
        $this->lastRefillTime = microtime(true);
        $this->mutex = fopen('php://memory', 'r+');
    }
    /**
     * 获取令牌（阻塞模式）
     * @param int $requested 需要的令牌数
     * @param float $timeout 最大等待时间（秒），0表示不等待
     * @return bool 是否成功获取
     */
    public function acquireBlocking(int $requested = 1, float $timeout = 0): bool
    {
        $startTime = microtime(true);
        while (true) {
            $this->lock();
            $this->refill();
            if ($this->tokens >= $requested) {
                $this->tokens -= $requested;
                $this->unlock();
                return true;
            }
            $this->unlock();
            // 检查超时
            if ($timeout > 0 && (microtime(true) - $startTime) >= $timeout) {
                return false;
            }
            // 计算需要等待的时间
            $needed = $requested - $this->tokens;
            $waitTime = $needed / $this->rate;
            // 短暂等待
            usleep(min(100000, $waitTime * 1000000));
        }
    }
    /**
     * 非阻塞获取令牌
     */
    public function acquireNonBlocking(int $requested = 1): bool
    {
        $this->lock();
        $this->refill();
        if ($this->tokens >= $requested) {
            $this->tokens -= $requested;
            $this->unlock();
            return true;
        }
        $this->unlock();
        return false;
    }
    /**
     * 获取等待时间（秒）
     */
    public function getWaitTime(int $requested = 1): float
    {
        $this->lock();
        $this->refill();
        if ($this->tokens >= $requested) {
            $this->unlock();
            return 0;
        }
        $needed = $requested - $this->tokens;
        $waitTime = $needed / $this->rate;
        $this->unlock();
        return $waitTime;
    }
    private function lock(): void
    {
        // 使用flock实现简单的进程内锁
        flock($this->mutex, LOCK_EX);
    }
    private function unlock(): void
    {
        flock($this->mutex, LOCK_UN);
    }
    private function refill(): void
    {
        $now = microtime(true);
        $elapsed = $now - $this->lastRefillTime;
        $newTokens = $elapsed * $this->rate;
        if ($newTokens > 0) {
            $this->tokens = min($this->capacity, $this->tokens + $newTokens);
            $this->lastRefillTime = $now;
        }
    }
    public function __destruct()
    {
        if (is_resource($this->mutex)) {
            fclose($this->mutex);
        }
    }
}

Redis分布式版本

<?php
/**
 * Redis实现分布式令牌桶
 */
class RedisTokenBucket
{
    private $redis;
    private $key;           // 令牌桶的key
    private $capacity;      // 桶容量
    private $rate;          // 令牌生成速率
    public function __construct($redis, string $key, int $capacity, float $rate)
    {
        $this->redis = $redis;
        $this->key = $key;
        $this->capacity = $capacity;
        $this->rate = $rate;
    }
    /**
     * 使用Lua脚本保证原子性
     */
    public function acquire(int $requested = 1): bool
    {
        $lua = <<<LUA
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local requested = tonumber(ARGV[3])
local now = tonumber(ARGV[4])
-- 获取当前令牌数和最后补充时间
local tokens = tonumber(redis.call('get', key .. ':tokens') or capacity)
local last_time = tonumber(redis.call('get', key .. ':time') or now)
-- 计算补充的令牌
local elapsed = now - last_time
local new_tokens = elapsed * rate
tokens = math.min(capacity, tokens + new_tokens)
-- 更新最后补充时间
redis.call('set', key .. ':time', now)
-- 检查是否有足够的令牌
if tokens >= requested then
    tokens = tokens - requested
    redis.call('set', key .. ':tokens', tokens)
    return 1
else
    redis.call('set', key .. ':tokens', tokens)
    return 0
end
LUA;
        $result = $this->redis->eval(
            $lua,
            [$this->key, $this->capacity, $this->rate, $requested, microtime(true)],
            1
        );
        return (bool)$result;
    }
    /**
     * 获取当前状态
     */
    public function getStatus(): array
    {
        return [
            'tokens' => (float)$this->redis->get($this->key . ':tokens') ?: $this->capacity,
            'last_time' => (float)$this->redis->get($this->key . ':time') ?: microtime(true),
        ];
    }
    /**
     * 重置令牌桶
     */
    public function reset(): void
    {
        $this->redis->del($this->key . ':tokens', $this->key . ':time');
    }
}

装饰器模式集成到业务代码

<?php
/**
 * 限流器接口
 */
interface RateLimiterInterface
{
    public function tryAcquire(int $requests = 1): bool;
    public function acquire(int $requests = 1, float $timeout = 0): bool;
}
/**
 * 令牌桶限流器（合并版本）
 */
class TokenBucketLimiter implements RateLimiterInterface
{
    private $bucket;
    public function __construct(int $capacity = 100, float $rate = 10)
    {
        $this->bucket = new AdvancedTokenBucket($capacity, $rate);
    }
    public function tryAcquire(int $requests = 1): bool
    {
        return $this->bucket->acquireNonBlocking($requests);
    }
    public function acquire(int $requests = 1, float $timeout = 0): bool
    {
        return $this->bucket->acquireBlocking($requests, $timeout);
    }
}
/**
 * 限流装饰器
 */
class RateLimitedService
{
    private $service;
    private $limiter;
    public function __construct($service, RateLimiterInterface $limiter)
    {
        $this->service = $service;
        $this->limiter = $limiter;
    }
    public function __call($method, $arguments)
    {
        // 尝试获取令牌
        if (!$this->limiter->tryAcquire()) {
            throw new \RuntimeException('请求过于频繁，请稍后再试', 429);
        }
        // 调用实际的服务方法
        return call_user_func_array([$this->service, $method], $arguments);
    }
}

使用示例

<?php
// 示例1：基本使用
$bucket = new TokenBucket(100, 10); // 容量100，每秒生成10个令牌
// 模拟请求
for ($i = 0; $i < 200; $i++) {
    $hasToken = $bucket->acquire();
    if ($hasToken) {
        echo "请求 $i: 通过\n";
    } else {
        echo "请求 $i: 被限流\n";
    }
    usleep(100000); // 延迟0.1秒
}
// 示例2：使用高级特性
$advancedBucket = new AdvancedTokenBucket(10, 5); // 容量10，每秒5个
// 非阻塞模式
$result = $advancedBucket->acquireNonBlocking(3);
echo $result ? "成功获取3个令牌" : "获取失败\n";
// 阻塞模式（等待最多2秒）
$result = $advancedBucket->acquireBlocking(5, 2.0);
if ($result) {
    echo "等待后成功获取5个令牌\n";
} else {
    echo "等待超时\n";
}
// 示例3：Redis分布式限流
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$limiter = new RedisTokenBucket($redis, 'api:user:123', 100, 10);
// 模拟并发请求
for ($i = 0; $i < 150; $i++) {
    if ($limiter->acquire()) {
        // 处理请求
    } else {
        // 返回429错误
    }
}
// 示例4：集成到Web应用
$service = new RateLimitedService(
    new UserService(),
    new TokenBucketLimiter(100, 20)
);
try {
    $result = $service->getUserInfo(1);
} catch (\RuntimeException $e) {
    if ($e->getCode() === 429) {
        header('HTTP/1.1 429 Too Many Requests');
        echo '请稍后再试';
    }
}

配置文件

<?php
return [
    'rate_limiter' => [
        'default' => [
            'driver' => 'redis',  // redis | memory
            'capacity' => 1000,
            'rate' => 100,        // 每秒
            'redis' => [
                'connection' => 'default',
            ],
        ],
        'api' => [
            'driver' => 'redis',
            'capacity' => 100,
            'rate' => 10,
        ],
        'login' => [
            'driver' => 'memory',
            'capacity' => 10,
            'rate' => 1,
        ],
    ],
];

测试代码

<?php
/**
 * 令牌桶单元测试
 */
class TokenBucketTest
{
    public function testBasicAcquire()
    {
        $bucket = new TokenBucket(10, 10);
        // 初始应该有10个令牌
        assert($bucket->acquire(10) === true);
        assert($bucket->acquire(1) === false);
        // 等待1秒后应该补充10个令牌
        sleep(1);
        assert($bucket->acquire(10) === true);
        echo "基础测试通过\n";
    }
    public function testRateLimiting()
    {
        $bucket = new TokenBucket(5, 1);
        $startTime = microtime(true);
        $successCount = 0;
        for ($i = 0; $i < 10; $i++) {
            if ($bucket->acquire()) {
                $successCount++;
            }
            usleep(200000);
        }
        $elapsed = microtime(true) - $startTime;
        // 2秒内应该能获取大约5-6个令牌（初始5个+补充）
        assert($successCount >= 5 && $successCount <= 6);
        echo "限流测试通过\n";
    }
    public function testBlockingAcquire()
    {
        $bucket = new AdvancedTokenBucket(1, 1);
        // 第一个立即成功
        assert($bucket->acquireBlocking(1, 0) === true);
        // 第二个等待1秒后成功
        $startTime = microtime(true);
        assert($bucket->acquireBlocking(1, 2) === true);
        $elapsed = microtime(true) - $startTime;
        assert($elapsed >= 0.9 && $elapsed <= 1.1);
        echo "阻塞等待测试通过\n";
    }
}
// 运行测试
$test = new TokenBucketTest();
$test->testBasicAcquire();
$test->testRateLimiting();
$test->testBlockingAcquire();

这个实现包含了：

  1. 基础单机版 - 适用于单进程应用
  2. 高级版 - 支持阻塞等待和并发控制
  3. Redis分布式版 - 适用于多机部署
  4. 业务集成 - 装饰器模式方便集成
  5. 完整测试 - 确保正确性

你可以根据实际需求选择合适版本使用。

上一篇PHP Lua 脚本扣库存

下一篇PHP 先查后更安全吗

相关文章

抱歉，评论功能暂时关闭!