本文目录导读：
在PHP项目中实现分布式锁,主要有以下几种常用方法，我会从简单到复杂依次介绍：
基于Redis的分布式锁（最常用）
简单实现
class RedisDistributedLock {
private $redis;
private $lockKey;
private $lockValue;
private $ttl;
public function __construct($redis, $lockKey, $ttl = 10) {
$this->redis = $redis;
$this->lockKey = $lockKey;
$this->lockValue = uniqid('', true); // 唯一标识
$this->ttl = $ttl;
}
// 获取锁
public function acquire() {
// SET NX EX 原子操作
$result = $this->redis->set(
$this->lockKey,
$this->lockValue,
['NX', 'EX' => $this->ttl]
);
return $result === true || $result === 'OK';
}
// 释放锁（使用Lua脚本保证原子性）
public function release() {
$script = <<<LUA
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
else
return 0
end
LUA;
$result = $this->redis->eval($script, [$this->lockKey, $this->lockValue], 1);
return $result === 1;
}
}
// 使用示例
$lock = new RedisDistributedLock($redis, 'order_lock:123', 30);
if ($lock->acquire()) {
try {
// 执行临界区代码
processOrder();
} finally {
$lock->release();
}
}
RedLock算法（高可用实现）
class RedLock {
private $redisInstances = [];
private $quorum;
private $retryDelay = 200; // 毫秒
private $retryCount = 3;
public function __construct(array $redisInstances) {
$this->redisInstances = $redisInstances;
$this->quorum = count($redisInstances) / 2 + 1;
}
public function lock($resource, $ttl = 1000) {
$token = uniqid('', true);
$startTime = microtime(true) * 1000;
for ($i = 0; $i < $this->retryCount; $i++) {
$successCount = 0;
$elapsedTime = microtime(true) * 1000 - $startTime;
foreach ($this->redisInstances as $instance) {
if ($elapsedTime >= $ttl) break;
$result = $instance->set(
$resource,
$token,
['NX', 'PX' => $ttl]
);
if ($result) {
$successCount++;
}
}
$elapsedTime = microtime(true) * 1000 - $startTime;
if ($successCount >= $this->quorum && $elapsedTime < $ttl) {
return ['token' => $token, 'resource' => $resource];
}
// 释放已获取的锁
$this->unlockInternal($resource, $token);
// 等待重试
usleep($this->retryDelay * 1000);
}
return false;
}
public function unlock($lockData) {
$this->unlockInternal($lockData['resource'], $lockData['token']);
}
private function unlockInternal($resource, $token) {
$script = <<<LUA
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
else
return 0
end
LUA;
foreach ($this->redisInstances as $instance) {
$instance->eval($script, [$resource, $token], 1);
}
}
}
基于数据库的分布式锁
MySQL行级锁实现
class MysqlDistributedLock {
private $pdo;
private $lockTable = 'distributed_locks';
public function __construct($pdo) {
$this->pdo = $pdo;
}
public function acquire($lockName, $timeout = 10) {
try {
// 使用GET_LOCK函数（MySQL 5.7+）
$stmt = $this->pdo->prepare("SELECT GET_LOCK(:lockName, :timeout)");
$stmt->execute([
':lockName' => $lockName,
':timeout' => $timeout
]);
return $stmt->fetchColumn() === 1;
} catch (PDOException $e) {
return false;
}
}
public function release($lockName) {
$stmt = $this->pdo->prepare("SELECT RELEASE_LOCK(:lockName)");
$stmt->execute([':lockName' => $lockName]);
return $stmt->fetchColumn() === 1;
}
}
基于数据库行锁的实现
class DatabaseRowLock {
private $pdo;
private $lockTable = 'locks';
public function acquire($lockName, $ttl = 30) {
$token = uniqid('', true);
try {
// 使用悲观锁
$this->pdo->beginTransaction();
// 查询并锁定行
$stmt = $this->pdo->prepare(
"SELECT * FROM {$this->lockTable}
WHERE lock_name = :lockName
AND (expires_at IS NULL OR expires_at < NOW())
FOR UPDATE"
);
$stmt->execute([':lockName' => $lockName]);
if ($stmt->rowCount() > 0) {
// 更新锁信息
$updateStmt = $this->pdo->prepare(
"UPDATE {$this->lockTable}
SET token = :token,
expires_at = DATE_ADD(NOW(), INTERVAL :ttl SECOND),
locked_at = NOW()
WHERE lock_name = :lockName"
);
$updateStmt->execute([
':token' => $token,
':ttl' => $ttl,
':lockName' => $lockName
]);
$this->pdo->commit();
return $token;
}
// 插入新锁
$insertStmt = $this->pdo->prepare(
"INSERT INTO {$this->lockTable} (lock_name, token, locked_at, expires_at)
VALUES (:lockName, :token, NOW(), DATE_ADD(NOW(), INTERVAL :ttl SECOND))"
);
$insertStmt->execute([
':lockName' => $lockName,
':token' => $token,
':ttl' => $ttl
]);
$this->pdo->commit();
return $token;
} catch (Exception $e) {
$this->pdo->rollBack();
return false;
}
}
public function release($lockName, $token) {
$stmt = $this->pdo->prepare(
"DELETE FROM {$this->lockTable}
WHERE lock_name = :lockName AND token = :token"
);
$stmt->execute([
':lockName' => $lockName,
':token' => $token
]);
return $stmt->rowCount() > 0;
}
}
基于ZooKeeper的分布式锁
class ZooKeeperLock {
private $zk;
private $root = '/locks';
public function __construct($hosts) {
$this->zk = new ZooKeeper($hosts);
// 确保根节点存在
if (!$this->zk->exists($this->root)) {
$this->zk->create($this->root, '', [
'flags' => ZooKeeper::EPHEMERAL
]);
}
}
public function acquire($lockName, $timeout = 10) {
$lockPath = $this->root . '/' . $lockName;
// 创建临时顺序节点
$createdPath = $this->zk->create($lockPath . '/lock-', '', [
'flags' => ZooKeeper::EPHEMERAL | ZooKeeper::SEQUENCE
]);
$startTime = time();
while (true) {
// 获取所有子节点
$children = $this->zk->getChildren($lockPath);
sort($children);
// 检查是否是第一个
$nodeName = basename($createdPath);
if ($nodeName === $children[0]) {
return $createdPath; // 获取锁成功
}
// 监听前一个节点
$prevNode = $children[array_search($nodeName, $children) - 1];
// 检查超时
if (time() - $startTime > $timeout) {
// 删除创建的节点
$this->zk->delete($createdPath);
return false; // 获取锁超时
}
// 等待直到前一个节点被删除
$this->zk->get($lockPath . '/' . $prevNode, function() {
// 前一个节点被删除，重新检查
});
}
}
public function release($lockPath) {
return $this->zk->delete($lockPath);
}
}
基于etcd的分布式锁
use Etcd\Client;
class EtcdLock {
private $client;
private $ttl = 10;
public function __construct($endpoints) {
$this->client = new Client($endpoints);
}
public function acquire($lockName) {
$key = '/locks/' . $lockName;
$value = uniqid('', true);
// 使用租约（lease）
$lease = $this->client->grant($this->ttl);
try {
// 尝试创建锁
$this->client->put($key, $value, [
'lease' => $lease->ID,
'prevKV' => true
]);
return $value; // 获取锁成功
} catch (\Exception $e) {
// 锁已被占用
$lease->revoke();
return false;
}
}
public function release($lockName, $value) {
$key = '/locks/' . $lockName;
// 只有持有者才能释放
$response = $this->client->get($key);
if ($response && $response->kvs[0]->value === $value) {
$this->client->delete($key);
return true;
}
return false;
}
public function refresh($lockName, $value) {
$key = '/locks/' . $lockName;
$response = $this->client->get($key);
if ($response && $response->kvs[0]->value === $value) {
$this->client->put($key, $value, [
'lease' => $response->kvs[0]->lease
]);
return true;
}
return false;
}
}
推荐实践建议
选择建议
class LockFactory {
public static function create($type, $config) {
switch ($type) {
case 'redis':
return new RedisDistributedLock(
$config['redis'],
$config['lockKey'],
$config['ttl'] ?? 30
);
case 'mysql':
return new MysqlDistributedLock($config['pdo']);
case 'zookeeper':
return new ZooKeeperLock($config['hosts']);
case 'etcd':
return new EtcdLock($config['endpoints']);
default:
throw new InvalidArgumentException("Unsupported lock type");
}
}
}
使用模版方法模式
abstract class AbstractDistributedLock {
abstract public function acquire();
abstract public function release();
public function execute(callable $callback) {
if ($this->acquire()) {
try {
return $callback();
} finally {
$this->release();
}
}
throw new LockFailedException('Failed to acquire lock');
}
}
// 使用示例
$lock = new RedisDistributedLock($redis, 'process_lock:123', 30);
$result = $lock->execute(function() {
// 业务逻辑
return processImportantTask();
});
注意事项
- 锁超时处理：设置合理的TTL，避免死锁
- 锁粒度：尽量精细，只锁定必要资源
- 重试机制：获取锁失败时的重试策略
- 监控告警：监控锁的获取和释放情况
- 降级策略：缓存降级或服务降级
对于大多数PHP项目,推荐使用Redis实现分布式锁，既简单又高效，如果需要更高的可靠性，可以考虑RedLock算法或ZooKeeper方案。
