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

PHP 令牌桶限流实现

wen PHP项目 159

本文目录导读：

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. 完整测试 - 确保正确性

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

版权声明：NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播

上一篇PHP Lua 脚本扣库存

下一篇PHP 先查后更安全吗

相关文章

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

友情链接攻略窝 · 礼包窝 · 配置窝 · 换算窝 · 算法窝 · 查文窝 · 手册窝 · 问答窝 · 全文搜 · 手册库
相关站点礼包窝 · 查文窝