PHP项目分布式信号量如何控制并发数量上限

本文目录导读：

1. Redis 分布式信号量
2. 基于 Redis 的令牌桶算法
3. 基于数据库的行锁
4. 使用消息队列控制并发
5. 分布式信号量组件
6. 推荐方案

在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();
        });
    }
}

推荐方案

1. 中小规模项目：使用 Redis 信号量，性能好，实现简单
2. 需要精确控制：令牌桶算法 + Redis
3. 已有 Redis 集群：RedLock 或 Redisson 实现
4. 需要持久化：使用数据库行锁方案
5. 高并发场景：使用消息队列 + 工作池模式

选择合适的方案时要考虑：

• 业务对精确度的要求
• 系统规模和并发量
• 基础设施已有的组件
• 网络延迟和可用性要求