本文目录导读：
在PHP分布式项目中控制并发数量上限,常见方案有以下几种：
Redis 分布式信号量
基于 Redis SETNX 实现
class RedisSemaphore {
private $redis;
private $key;
private $maxConcurrency;
private $timeout;
public function __construct($redis, $key, $maxConcurrency, $timeout = 10) {
$this->redis = $redis;
$this->key = "semaphore:{$key}";
$this->maxConcurrency = $maxConcurrency;
$this->timeout = $timeout;
}
public function acquire($identifier = null) {
$identifier = $identifier ?: uniqid('', true);
$startTime = time();
while (time() - $startTime < $this->timeout) {
// 使用 Lua 脚本保证原子性
$script = <<<LUA
local current = redis.call('LLEN', KEYS[1])
if current < tonumber(ARGV[1]) then
redis.call('RPUSH', KEYS[1], ARGV[2])
redis.call('EXPIRE', KEYS[1], ARGV[3])
return 1
end
return 0
LUA;
$result = $this->redis->eval($script,
[$this->key, $this->maxConcurrency, $identifier, $this->timeout],
1
);
if ($result) {
return $identifier;
}
// 等待重试
usleep(100000); // 100ms
}
return false;
}
public function release($identifier) {
$script = <<<LUA
local items = redis.call('LRANGE', KEYS[1], 0, -1)
for i, v in ipairs(items) do
if v == ARGV[1] then
redis.call('LREM', KEYS[1], 1, ARGV[1])
return 1
end
end
return 0
LUA;
return $this->redis->eval($script, [$this->key, $identifier], 1);
}
}
// 使用示例
$semaphore = new RedisSemaphore($redis, 'api_limit', 10);
$identifier = $semaphore->acquire();
if ($identifier) {
try {
// 执行需要限制并发的代码
processData();
} finally {
$semaphore->release($identifier);
}
} else {
throw new \Exception('并发上限已达到，请稍后重试');
}
基于 Redis 的令牌桶算法
class TokenBucket {
private $redis;
private $key;
private $capacity;
private $rate; // 每秒添加的令牌数
public function __construct($redis, $key, $capacity, $rate) {
$this->redis = $redis;
$this->key = "token_bucket:{$key}";
$this->capacity = $capacity;
$this->rate = $rate;
}
public function consume($tokens = 1) {
$script = <<<LUA
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local tokens = tonumber(ARGV[3])
local now = tonumber(ARGV[4])
local last_time = redis.call('GET', key .. ':time')
local current_tokens
if last_time then
last_time = tonumber(last_time)
local elapsed = now - last_time
current_tokens = tonumber(redis.call('GET', key .. ':tokens') or 0)
current_tokens = math.min(capacity, current_tokens + elapsed * rate)
else
current_tokens = capacity
end
if current_tokens >= tokens then
redis.call('SET', key .. ':tokens', current_tokens - tokens)
redis.call('SET', key .. ':time', now)
return 1
end
return 0
LUA;
return $this->redis->eval($script, [
$this->key,
$this->capacity,
$this->rate,
$tokens,
time()
], 1);
}
}
基于数据库的行锁
class DBSemaphore {
private $db;
private $table = 'semaphores';
public function acquire($name, $maxCount, $timeout = 30) {
$startTime = time();
while (time() - $startTime < $timeout) {
try {
$this->db->beginTransaction();
// 获取当前并发数（使用行锁）
$row = $this->db->select("SELECT * FROM {$this->table} WHERE name = ? FOR UPDATE", [$name]);
if (!$row) {
$this->db->insert($this->table, [
'name' => $name,
'count' => 1,
'max_count' => $maxCount
]);
$this->db->commit();
return true;
}
if ($row->count < $maxCount) {
$this->db->update($this->table,
['count' => $row->count + 1],
['name' => $name]
);
$this->db->commit();
return true;
}
$this->db->rollback();
} catch (\Exception $e) {
$this->db->rollback();
}
usleep(100000); // 100ms
}
return false;
}
public function release($name) {
$this->db->beginTransaction();
$row = $this->db->select("SELECT * FROM {$this->table} WHERE name = ? FOR UPDATE", [$name]);
if ($row && $row->count > 0) {
$this->db->update($this->table,
['count' => $row->count - 1],
['name' => $name]
);
}
$this->db->commit();
}
}
使用消息队列控制并发
class QueueBasedLimiter {
private $redis;
private $processKey = 'process:concurrent';
private $maxConcurrent = 10;
public function canProcess($taskId) {
$script = <<<LUA
local key = KEYS[1]
local max = tonumber(ARGV[1])
local task = ARGV[2]
local current = redis.call('SCARD', key)
if current < max then
redis.call('SADD', key, task)
return 1
end
return 0
LUA;
return $this->redis->eval($script,
[$this->processKey, $this->maxConcurrent, $taskId],
1
);
}
public function finishProcess($taskId) {
$this->redis->srem($this->processKey, $taskId);
}
// 消费者端
public function consumer() {
while (true) {
$task = $this->redis->lpop('task:queue');
if ($task && $this->canProcess($task)) {
try {
// 处理任务
processTask($task);
} finally {
$this->finishProcess($task);
}
} else {
usleep(100000);
}
}
}
}
分布式信号量组件
interface DistributedSemaphoreInterface {
public function acquire($key, $maxConcurrency, $timeout = 10);
public function release($key);
}
// 封装成 Trait 方便使用
trait SemaphoreTrait {
private $semaphore;
protected function withSemaphore($key, $maxConcurrency, callable $callback) {
$identifier = $this->semaphore->acquire($key, $maxConcurrency);
if (!$identifier) {
throw new \Exception("并发上限 {$maxConcurrency} 已达到");
}
try {
return $callback();
} finally {
$this->semaphore->release($key, $identifier);
}
}
}
class ApiService {
use SemaphoreTrait;
public function heavyOperation() {
return $this->withSemaphore('heavy_api', 5, function() {
// 限制最多5个并发
return $this->doHeavyWork();
});
}
}
推荐方案
- 中小规模项目：使用 Redis 信号量，性能好，实现简单
- 需要精确控制：令牌桶算法 + Redis
- 已有 Redis 集群：RedLock 或 Redisson 实现
- 需要持久化：使用数据库行锁方案
- 高并发场景：使用消息队列 + 工作池模式
选择合适的方案时要考虑：
- 业务对精确度的要求
- 系统规模和并发量
- 基础设施已有的组件
- 网络延迟和可用性要求
