kode/limiting

高性能 PHP 限流器，支持令牌桶、滑动窗口算法，可用于本地和分布式限流。

功能特性

令牌桶算法：支持突发流量，按固定速率补充令牌
滑动窗口算法：基于时间窗口的精确限流
本地限流：使用内存存储，适用于单进程
分布式限流：使用 Redis，支持单机/Sentinel/Cluster 模式
并发控制：支持任务、进程、Fiber 协程的本地和分布式限流
原子操作：Lua 脚本保证分布式环境下的原子性
高可用：支持 Redis Sentinel 和 Cluster 自动故障转移
PHP 8.2+：使用 readonly、enum 等新特性优化性能

系统要求

PHP >= 8.2
可选：Redis 扩展（用于分布式部署）

安装

composer require kode/limiting

统一入口（推荐）

使用 Limiter 类可以更简洁地创建各种限流器：

use Kode\Limiting\Limiter;

// 令牌桶限流器（默认）
$limiter = Limiter::tokenBucket(100, 10.0);
$limiter->allow('api:user:123');

// 滑动窗口限流器
$limiter = Limiter::slidingWindow(100, 60.0);

// 漏桶限流器
$limiter = Limiter::leakyBucket(100, 1.0);

// 计数器限流器
$limiter = Limiter::counter(1000, 60);

// Redis 分布式限流器
$limiter = Limiter::redis(
    \Kode\Limiting\Enum\LimiterType::TOKEN_BUCKET,
    100, 10.0,
    '127.0.0.1', 6379
);

// Memcached 分布式限流器
$limiter = Limiter::memcached(
    \Kode\Limiting\Enum\LimiterType::TOKEN_BUCKET,
    100, 10.0,
    '127.0.0.1', 11211
);

// PDO 分布式限流器
$limiter = Limiter::pdo(
    \Kode\Limiting\Enum\LimiterType::TOKEN_BUCKET,
    100, 10.0,
    'mysql:host=127.0.0.1;dbname=limiting', 'root', 'password'
);

// 并发控制器
$taskLimiter = Limiter::task(10);      // 任务并发控制
$processLimiter = Limiter::process(10); // 进程并发控制
$fiberLimiter = Limiter::fiber(10);     // Fiber 并发控制

// 中间件
$middleware = Limiter::middleware(\Kode\Limiting\Enum\LimiterType::TOKEN_BUCKET, 100, 10.0);

快速开始

本地限流（令牌桶）

use Kode\Limiting\Algorithm\TokenBucket;
use Kode\Limiting\Store\MemoryStore;

$store = new MemoryStore();
$bucket = new TokenBucket($store, 100, 10.0);

if ($bucket->allow('user:123', 1)) {
    echo "请求通过";
} else {
    echo "被限流";
}

echo "剩余令牌: " . $bucket->getRemaining('user:123');
echo "等待时间: " . $bucket->getWaitTime('user:123') . "秒";

滑动窗口限流

use Kode\Limiting\Algorithm\SlidingWindow;
use Kode\Limiting\Store\MemoryStore;

$window = new SlidingWindow($store, 100, 1.0);

if ($window->allow('api:v1', 1)) {
    echo "请求通过";
}

分布式限流（跨机器）

use Kode\Limiting\Distributed\DistributedLimiter;

// 单机模式
$limiter = DistributedLimiter::create(
    '127.0.0.1', 6379,
    1000,    // 容量
    100.0    // 每秒补充100个令牌
);

// Sentinel 高可用模式
$limiter = DistributedLimiter::createSentinel(
    ['192.168.1.1:26379', '192.168.1.2:26379'],
    'mymaster',
    1000,
    100.0
);

// Cluster 分片模式
$limiter = DistributedLimiter::createCluster(
    ['192.168.1.1:6379', '192.168.1.2:6379', '192.168.1.3:6379'],
    1000,
    100.0
);

if ($limiter->allow('global:api', 1)) {
    echo "请求通过";
}

并发任务控制

use Kode\Limiting\Concurrency\TaskLimiter;

$limiter = TaskLimiter::create(10, 10, 1.0);

// 手动获取和释放
if ($limiter->tryAcquire('task:1')) {
    try {
        do_something();
    } finally {
        $limiter->release('task:1');
    }
}

// 自动释放
$result = $limiter->run('task:2', fn() => do_something());

// 阻塞等待
$limiter->acquire('task:3', 30.0);

架构设计

src/
├── Limiter.php                    # 统一入口类
├── DTO/                          # 数据传输对象
│   ├── LimiterConfig.php        # 限流器配置（readonly）
│   └── LimiterResult.php        # 限流结果（readonly）
├── Enum/                         # 枚举类型
│   ├── LimiterType.php          # 限流器类型
│   ├── StoreType.php            # 存储类型
│   └── RedisMode.php           # Redis 模式
├── Store/                        # 存储层
│   ├── StoreInterface.php       # 存储接口
│   ├── MemoryStore.php          # 内存存储
│   ├── RedisStore.php           # Redis 存储（支持 Sentinel/Cluster）
│   ├── MemcachedStore.php       # Memcached 存储
│   └── PdoStore.php            # PDO 存储（MySQL/SQLite/PostgreSQL）
├── Algorithm/                    # 限流算法
│   ├── RateLimiterInterface.php # 限流器接口
│   ├── TokenBucket.php          # 令牌桶
│   ├── SlidingWindow.php        # 滑动窗口
│   ├── LeakyBucket.php          # 漏桶
│   └── Counter.php              # 计数器
├── Distributed/                   # 分布式限流
│   ├── DistributedLimiter.php           # 分布式令牌桶
│   ├── DistributedTaskLimiter.php       # 分布式任务限流
│   ├── DistributedProcessLimiter.php    # 分布式进程限流
│   └── DistributedFiberLimiter.php      # 分布式 Fiber 限流
├── Concurrency/                   # 本地并发控制
│   ├── TaskLimiter.php           # 任务限流
│   ├── ProcessLimiter.php        # 进程限流
│   └── FiberLimiter.php         # Fiber 限流
└── Middleware/                    # 中间件
    ├── LimiterMiddleware.php          # 限流中间件
    └── LimiterMiddlewareInterface.php # 中间件接口

API 文档

StoreInterface

存储接口，所有存储实现必须实现此接口。

public function get(string $key): ?string;              // 获取值
public function set(string $key, string $value, int $ttl = 0): void;  // 设置值
public function delete(string $key): void;             // 删除键
public function incr(string $key, int $step = 1): int; // 原子递增
public function ttl(string $key): int;                 // 获取 TTL

RateLimiterInterface

限流器接口。

public function allow(string $key, int $tokens = 1): bool;  // 检查是否允许
public function getRemaining(string $key): float;            // 获取剩余数量
public function getWaitTime(string $key): float;             // 获取等待时间
public function reset(string $key): void;                    // 重置

Limiter（统一入口）

统一入口类，提供简洁的 API 创建限流器。

// 工厂方法
Limiter::tokenBucket(int $capacity, float $refillRate, $store);      // 令牌桶
Limiter::slidingWindow(int $capacity, float $windowSize, $store);    // 滑动窗口
Limiter::leakyBucket(int $capacity, float $leakRate, $store);       // 漏桶
Limiter::counter(int $maxRequests, int $windowSeconds, $store);      // 计数器
Limiter::redis(LimiterType, int $capacity, float $refillRate, ...); // Redis 限流
Limiter::memcached(LimiterType, int $capacity, float $refillRate, ...); // Memcached 限流
Limiter::pdo(LimiterType, int $capacity, float $refillRate, ...);    // PDO 限流
Limiter::task(int $maxConcurrency);    // 任务并发控制器
Limiter::process(int $maxConcurrency); // 进程并发控制器
Limiter::fiber(int $maxConcurrency);  // Fiber 并发控制器
Limiter::middleware(LimiterType, int $capacity, float $refillRate);  // 中间件

// 实例方法
$limiter->allow(string $key, int $tokens);   // 检查是否允许
$limiter->getRemaining(string $key);         // 获取剩余数量
$limiter->getWaitTime(string $key);          // 获取等待时间
$limiter->reset(string $key);                // 重置
$limiter->build();                           // 获取底层限流器
$limiter->getStore();                        // 获取存储
$limiter->getConfig();                       // 获取配置

TokenBucket

令牌桶限流算法。

$bucket = new TokenBucket($store, $capacity, $refillRate, $ttl = 3600, $prefix = 'bucket:');

$bucket->allow('key', 1);           // 检查是否允许
$bucket->getRemaining('key');        // 获取剩余令牌数
$bucket->getWaitTime('key');         // 获取等待时间
$bucket->reset('key');               // 重置
$bucket->getCapacity();              // 获取容量
$bucket->getRefillRate();           // 获取补充速率
$bucket->getTtl();                  // 获取 TTL
$bucket->check('key', 1);           // 检查并返回 LimiterResult

SlidingWindow

滑动窗口限流算法。

$window = new SlidingWindow($store, $capacity, $windowSize = 1.0, $ttl = 3600, $prefix = 'sw:');

$window->allow('key', 1);           // 检查是否允许
$window->getRemaining('key');        // 获取剩余请求数
$window->getWaitTime('key');         // 获取等待时间
$window->reset('key');               // 重置
$window->getCapacity();               // 获取容量
$window->getWindowSize();            // 获取窗口大小
$window->check('key', 1);           // 检查并返回 LimiterResult

DistributedLimiter

分布式限流器（基于 Redis）。

// 单机模式
$limiter = DistributedLimiter::create($host, $port, $capacity, $refillRate, $password, $database);

// Sentinel 模式
$limiter = DistributedLimiter::createSentinel($sentinels, $masterName, $capacity, $refillRate);

// Cluster 模式
$limiter = DistributedLimiter::createCluster($nodes, $capacity, $refillRate);

// 方法
$limiter->allow('key', 1);                    // 检查是否允许（原子操作）
$limiter->tryAcquire('key', 1);               // 尝试获取（别名）
$limiter->getRemaining('key');                 // 获取剩余令牌数
$limiter->getWaitTime('key');                  // 获取等待时间
$limiter->reset('key');                        // 重置
$limiter->allowBatch(['key1', 'key2'], 1);     // 批量检查
$limiter->check('key', 1);                     // 检查并返回 LimiterResult

分布式并发控制器

use Kode\Limiting\Distributed\DistributedTaskLimiter;
use Kode\Limiting\Distributed\DistributedProcessLimiter;
use Kode\Limiting\Distributed\DistributedFiberLimiter;

$taskLimiter = DistributedTaskLimiter::create('127.0.0.1', 6379, 10);
$processLimiter = DistributedProcessLimiter::create('127.0.0.1', 6379, 10);
$fiberLimiter = DistributedFiberLimiter::create('127.0.0.1', 6379, 100);

$taskLimiter->tryAcquire('task:1');           // 非阻塞获取
$taskLimiter->acquire('task:1', 30.0);        // 阻塞等待
$taskLimiter->run('task:1', fn() => null);   // 执行任务
$taskLimiter->release('task:1');               // 释放
$taskLimiter->getActiveCount();                // 当前活跃数

本地并发控制器

use Kode\Limiting\Concurrency\TaskLimiter;
use Kode\Limiting\Concurrency\ProcessLimiter;
use Kode\Limiting\Concurrency\FiberLimiter;

$taskLimiter = TaskLimiter::create($maxConcurrency, $capacity, $refillRate);
$processLimiter = ProcessLimiter::getInstance($maxProcesses, $capacity, $refillRate);
$fiberLimiter = new FiberLimiter($maxFibers, $capacity, $refillRate, $store);

$taskLimiter->tryAcquire('task:1');
$taskLimiter->acquire('task:1', 30.0);
$taskLimiter->run('task:1', fn() => null);
$taskLimiter->release('task:1');
$taskLimiter->getMaxConcurrency();

DTO 对象

LimiterConfig

限流器配置（不可变对象）。

$config = new LimiterConfig($capacity, $refillRate, $ttl = 3600, $prefix = 'limiter:');

$config->capacity;      // 容量
$config->refillRate;    // 补充速率
$config->ttl;           // TTL
$config->prefix;        // 前缀

// 链式调用创建新实例
$newConfig = $config->withCapacity(200)->withRefillRate(20.0);

LimiterResult

限流结果（不可变对象）。

$result = $bucket->check('key', 1);

$result->allowed;       // 是否允许
$result->remaining;    // 剩余数量
$result->waitTime;     // 等待时间
$result->timestamp;    // 时间戳

$result->isAllowed();   // 是否允许（布尔值）
$result->toArray();     // 转换为数组

枚举类型

use Kode\Limiting\Enum\LimiterType;
use Kode\Limiting\Enum\StoreType;
use Kode\Limiting\Enum\RedisMode;

LimiterType::TOKEN_BUCKET->label();      // 令牌桶
LimiterType::SLIDING_WINDOW->label();    // 滑动窗口

StoreType::MEMORY->label();              // 内存存储
StoreType::REDIS->label();               // Redis 存储

RedisMode::STANDALONE->label();          // 单机模式
RedisMode::SENTINEL->label();            // Sentinel 高可用
RedisMode::CLUSTER->label();             // Cluster 分片

单元测试

测试覆盖所有核心功能，共计 35 个测试用例。

测试用例列表

Token Bucket (令牌桶)
 ✔ 允许首次请求
 ✔ 允许多次请求
 ✔ 超出容量拒绝
 ✔ 获取剩余令牌数
 ✔ 新键返回完整容量
 ✔ 获取等待时间
 ✔ 重置限流器
 ✔ 时间推移后补充令牌

Sliding Window (滑动窗口)
 ✔ 容量内允许通过
 ✔ 超出容量拒绝
 ✔ 获取剩余请求数
 ✔ 请求后剩余数正确
 ✔ 重置限流器
 ✔ 可用时等待时间为0
 ✔ 耗尽时等待时间大于0

Memory Store (内存存储)
 ✔ 设置和获取
 ✔ 获取不存在的键
 ✔ 删除键
 ✔ 递增操作
 ✔ 新键递增
 ✔ TTL 计算
 ✔ 清空存储

Task Limiter (任务限流)
 ✔ 尝试获取许可
 ✔ 多次获取
 ✔ 释放许可
 ✔ 执行任务
 ✔ 获取最大并发数

Process Limiter (进程限流)
 ✔ 尝试获取许可
 ✔ 释放许可
 ✔ 获取最大进程数
 ✔ 单例模式

Fiber Limiter (Fiber 限流)
 ✔ 支持检查
 ✔ 尝试获取许可
 ✔ 释放许可
 ✔ 获取最大 Fiber 数

运行测试

# 运行所有测试
./vendor/bin/phpunit

# 显示详细输出
./vendor/bin/phpunit --testdox

# 运行指定测试类
./vendor/bin/phpunit tests/TokenBucketTest.php

# 生成覆盖率报告
./vendor/bin/phpunit --coverage-html coverage

自定义存储

实现 StoreInterface 接口即可：

use Kode\Limiting\Store\StoreInterface;

class MyStore implements StoreInterface
{
    public function get(string $key): ?string { /* ... */ }
    public function set(string $key, string $value, int $ttl = 0): void { /* ... */ }
    public function delete(string $key): void { /* ... */ }
    public function incr(string $key, int $step = 1): int { /* ... */ }
    public function ttl(string $key): int { /* ... */ }
}

Redis 配置

单机模式

$limiter = DistributedLimiter::create(
    '127.0.0.1',    // 主机
    6379,           // 端口
    1000,           // 容量
    100.0,          // 补充速率
    'password',     // 密码（可选）
    0               // 数据库编号
);

Sentinel 模式

$limiter = DistributedLimiter::createSentinel(
    ['192.168.1.1:26379', '192.168.1.2:26379', '192.168.1.3:26379'],
    'mymaster',     // 主节点名称
    1000,
    100.0,
    'sentinel_pass'
);

Cluster 模式

$limiter = DistributedLimiter::createCluster(
    ['192.168.1.1:6379', '192.168.1.2:6379', '192.168.1.3:6379'],
    1000,
    100.0,
    'cluster_pass'
);

使用场景

API 限流

$limiter = DistributedLimiter::create('127.0.0.1', 6379, 100, 10.0);

$response = $limiter->allow('api:user:' . $userId, 1)
    ? processRequest()
    : new Response('Rate limited', 429);

秒杀活动

$limiter = DistributedLimiter::create('127.0.0.1', 6379, 1000, 0);

if (!$limiter->allow('seckill:' . $productId, 1)) {
    throw new Exception('活动已结束');
}

并发控制

$limiter = DistributedTaskLimiter::create('127.0.0.1', 6379, 10);

foreach ($tasks as $task) {
    $limiter->run('task:' . $task['id'], function () use ($task) {
        processTask($task);
    });
}

开发

# 安装依赖
composer install

# 运行测试
./vendor/bin/phpunit

# 语法检查
find src -name "*.php" -exec php -l {} \;

目录结构

kode/limiting/
├── src/
│   ├── DTO/                       # 数据传输对象
│   ├── Enum/                      # 枚举类型
│   ├── Store/                     # 存储层
│   ├── Algorithm/                 # 限流算法
│   ├── Distributed/              # 分布式限流
│   └── Concurrency/               # 并发控制
├── tests/                         # 单元测试
├── composer.json
├── phpunit.xml
├── LICENSE                        # Apache 2.0
└── README.md

许可证

Apache License 2.0 - 参见 LICENSE 文件

Name		Name	Last commit message	Last commit date
Latest commit History 9 Commits
src		src
tests		tests
.gitignore		.gitignore
LICENSE		LICENSE
README.md		README.md
composer.json		composer.json

Folders and files

Latest commit

History

Repository files navigation

kode/limiting

功能特性

系统要求

安装

统一入口（推荐）

快速开始

本地限流（令牌桶）

滑动窗口限流

分布式限流（跨机器）

并发任务控制

架构设计

API 文档

StoreInterface

RateLimiterInterface

Limiter（统一入口）

TokenBucket

SlidingWindow

DistributedLimiter

分布式并发控制器

本地并发控制器

DTO 对象

LimiterConfig

LimiterResult

枚举类型

单元测试

测试用例列表

运行测试

自定义存储

Redis 配置

单机模式

Sentinel 模式

Cluster 模式

使用场景

API 限流

秒杀活动

并发控制

开发

目录结构

许可证

About

Resources

License

Uh oh!

Stars

Watchers

Forks

Releases

Packages 0

Uh oh!

Contributors

Uh oh!

Languages

Packages