更多课程，请关注以上公众号

电商大促超卖、库存扣减竞态、重复下单……这些高并发经典噩梦，根源都指向同一个问题——分布式锁没搞好。本文手把手带你用 PHP + Swoole + Redis 构建一套零外部依赖的生产级分布式锁，性能对比数据揭示超卖的本质，附赠可直接上线的完整代码。

一、超卖的本质：你以为的"原子操作"根本不原子

先复现一下经典超卖场景。假设库存只剩 1 件，1000 个请求同时打过来：

<?php
// ❌ 经典超卖代码（反面教材）
functiondecreaseStock(int$productId, int$quantity): bool
{
$stock = Redis::get("stock:{$productId}");  // ① 查库存

if ($stock < $quantity) {
returnfalse; // 库存不足
    }

// ② 模拟业务处理（0.1ms）
usleep(100);

// ③ 扣减库存
Redis::decrby("stock:{$productId}", $quantity);

returntrue;
}

问题在哪？ 步骤 ① → ② → ③ 之间没有任何保护，1000 个并发请求会同时读到库存=1，同时认为「够扣」，然后同时执行扣减。最终库存变成 -999，超卖了 999 件。

这不是 PHP 的 bug，这是分布式系统的经典 TOCTOU（Time-of-Check-Time-of-Use）竞态条件。

1.1 超卖压测数据（先看结果，再看方案）

结论：MySQL 行锁能解决超卖，但吞吐量是瓶颈。Redis 分布式锁可以在保证正确性的同时把吞吐量拉高 15 倍。

二、分布式锁的三个核心要求（很多人只知道第一个）

在写代码之前，先搞清楚一把「好锁」必须满足的三个条件：

要求 1：互斥性（Mutual Exclusion）

任意时刻只有一个客户端持有锁。这是最基本的，用 Redis SET key value NX 就能做到。

要求 2：防死锁（Deadlock Prevention）

持有锁的进程崩溃后，锁必须自动释放。解决方案：给锁设置过期时间。

# 原子操作：SET key value NX EX seconds
SET lock:order:12345 "worker-1" NX EX 30

要求 3：只能自己释放（Ownership）

这是最容易被忽略的！ 进程 A 的锁，不能被进程 B 误释放。

// ❌ 危险写法：直接 DEL，可能删掉别人的锁
Redis::del("lock:order:12345");

// ✅ 安全写法：先验证是不是自己的锁，再删
// 但这里又有个竞态……需要用 Lua 脚本保证原子性

三、完整实现：生产级分布式锁

3.1 基础版：Redis 分布式锁

<?php

declare(strict_types=1);

namespaceApp\Lock;

useRedis;
useRuntimeException;

/**
 * Redis 分布式锁
 * 
 * 特性：
 * - 原子性加锁（SET NX EX）
 * - 唯一持有者标识（UUID）
 * - Lua 脚本原子性解锁
 * - 自动续期（可选）
 */
classRedisDistributedLock
{
private Redis $redis;

/** 锁的持有者标识（进程唯一） */
privatestring$owner;

/**
     * @param Redis  $redis      Redis 连接
     * @param string $prefix     锁 key 前缀
     * @param int    $ttl        锁过期时间（秒），防死锁
     * @param int    $retryTimes 加锁失败时重试次数
     * @param int    $retryDelay 重试间隔（毫秒）
     */
publicfunction__construct(
        Redis $redis,
privatereadonlystring$prefix = 'lock:',
privatereadonlyint$ttl = 30,
privatereadonlyint$retryTimes = 3,
privatereadonlyint$retryDelay = 100,
) {
$this->redis = $redis;
// 每个进程/协程生成唯一标识，防止误释放
$this->owner = sprintf(
'%s:%s:%s',
gethostname(),
getmypid(),
bin2hex(random_bytes(8))
        );
    }

/**
     * 尝试加锁
     *
     * @param string $resource 要锁定的资源名（如 "order:12345"）
     * @return bool
     */
public function acquire(string$resource): bool
{
$key = $this->prefix . $resource;

for ($attempt = 0; $attempt <= $this->retryTimes; $attempt++) {
// SET key value NX EX ttl — 原子操作，不存在才设置
$result = $this->redis->set(
$key,
$this->owner,
                ['NX', 'EX' => $this->ttl]
            );

if ($result !== false) {
returntrue; // 加锁成功
            }

if ($attempt < $this->retryTimes) {
// 随机退避，避免所有进程同时重试（惊群效应）
usleep(($this->retryDelay + random_int(0, 50)) * 1000);
            }
        }

returnfalse; // 加锁失败
    }

/**
     * 释放锁（Lua 脚本保证原子性）
     *
     * @param string $resource 资源名
     * @return bool
     */
public function release(string$resource): bool
{
$key = $this->prefix . $resource;

/**
         * Lua 脚本：先比较持有者，再删除
         * 整个操作原子执行，不会有竞态条件
         */
$lua = <<<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($lua, [$key, $this->owner], 1);

return (int) $result === 1;
    }

/**
     * 续期（延长锁的过期时间）
     * 用于业务处理时间不确定的场景
     *
     * @param string $resource 资源名
     * @param int    $ttl      新的 TTL（秒）
     * @return bool
     */
publicfunctionrenew(string$resource, int$ttl = 0): bool
{
$key = $this->prefix . $resource;
$ttl = $ttl ?: $this->ttl;

// Lua 脚本：只有持有者才能续期
$lua = <<<LUA
            if redis.call("GET", KEYS[1]) == ARGV[1] then
                return redis.call("EXPIRE", KEYS[1], ARGV[2])
            else
                return 0
            end
        LUA;

$result = $this->redis->eval($lua, [$key, $this->owner, (string) $ttl], 1);

return (int) $result === 1;
    }

/**
     * 带锁执行闭包（推荐使用这个，自动处理加/解锁）
     *
     * @param string   $resource 资源名
     * @param callable $callback 需要加锁执行的业务逻辑
     * @return mixed 闭包的返回值
     * @throws RuntimeException 加锁失败时抛出异常
     */
publicfunctionwithLock(string$resource, callable$callback): mixed
{
if (!$this->acquire($resource)) {
thrownewRuntimeException(
"Failed to acquire lock for resource: {$resource}"
            );
        }

try {
return$callback();
        } finally {
// 无论成功失败，都要释放锁
$this->release($resource);
        }
    }
}

3.2 实战应用：彻底消灭超卖

<?php

declare(strict_types=1);

namespaceApp\Service;

useApp\Lock\RedisDistributedLock;
useRedis;
useRuntimeException;

classStockService
{
private RedisDistributedLock $lock;
private Redis $redis;

publicfunction__construct(Redis $redis)
{
$this->redis = $redis;
$this->lock = newRedisDistributedLock(
            redis: $redis,
            prefix: 'lock:stock:',
            ttl: 5,           // 扣库存操作，5秒够了
            retryTimes: 3,    // 最多重试3次
            retryDelay: 50,   // 50ms 重试间隔
        );
    }

/**
     * 安全扣减库存（有分布式锁保护）
     *
     * @param int $productId 商品ID
     * @param int $quantity  扣减数量
     * @return bool
     */
publicfunctiondecreaseStockSafe(int$productId, int$quantity): bool
{
$resource = "product:{$productId}";

try {
return$this->lock->withLock($resource, function () use ($productId, $quantity) {
                // 临界区：同一时刻只有一个进程执行这里
                $stockKey = "stock:{$productId}";
$stock = (int) $this->redis->get($stockKey);

if ($stock < $quantity) {
returnfalse; // 库存不足，返回失败
                }

// 原子扣减
$this->redis->decrby($stockKey, $quantity);

// 记录扣减日志（可选）
$this->redis->lpush(
"stock:log:{$productId}",
json_encode([
'action' => 'deduct',
'quantity' => $quantity,
'time' => time(),
'stock_after' => $stock - $quantity,
                    ])
                );

returntrue;
            });
        } catch (RuntimeException$e) {
// 加锁失败（高并发下锁竞争激烈）
// 可以降级：写入消息队列，异步处理
logger()->warning('Lock acquire failed', [
'product_id' => $productId,
'error' => $e->getMessage(),
            ]);
returnfalse;
        }
    }
}

四、进阶：Swoole 协程版高性能分布式锁

上面的方案在传统 PHP-FPM 下已经够用了。但如果你用 Swoole / Hyperf / Workerman，可以用协程版本——非阻塞等待，性能更高。

4.1 Swoole 协程锁（非阻塞等待）

<?php

declare(strict_types=1);

namespaceApp\Lock;

useSwoole\Coroutine;
useSwoole\Coroutine\Channel;

/**
 * Swoole 协程分布式锁
 * 
 * 对比普通 Redis 锁的优势：
 * - 等锁时让出协程（非阻塞），不浪费 CPU
 * - 超时自动放弃，不无限等待
 * - 支持同进程内的协程级互斥
 */
classSwooleDistributedLock
{
/** 本地协程等待队列 key => Channel[] */
privatestaticarray$waitQueues = [];

publicfunction__construct(
privatereadonly \Redis $redis,
privatereadonlystring$prefix = 'swoole_lock:',
privatereadonlyint$ttl = 30,
privatereadonlyfloat$timeout = 3.0,  // 等待锁的最大时间（秒）
) {}

/**
     * 加锁（协程安全，支持超时）
     */
publicfunctionacquire(string$resource): bool
{
$key = $this->prefix . $resource;
$owner = $this->generateOwner();
$deadline = microtime(true) + $this->timeout;

while (microtime(true) < $deadline) {
// 尝试加锁
$result = $this->redis->set($key, $owner, ['NX', 'EX' => $this->ttl]);

if ($result !== false) {
// 存到协程上下文，释放时用
Coroutine::getContext()["lock:{$resource}"] = $owner;
returntrue;
            }

// 加锁失败，让出协程，等待 10ms 后重试
// Swoole 协程让出不会阻塞进程，其他协程照常运行
Coroutine::sleep(0.01);
        }

returnfalse; // 超时
    }

/**
     * 释放锁
     */
publicfunctionrelease(string$resource): bool
{
$key = $this->prefix . $resource;
$owner = Coroutine::getContext()["lock:{$resource}"] ?? null;

if ($owner === null) {
returnfalse; // 当前协程没有这把锁
        }

$lua = <<<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($lua, [$key, $owner], 1);
unset(Coroutine::getContext()["lock:{$resource}"]);

return (int) $result === 1;
    }

privatefunctiongenerateOwner(): string
{
returnsprintf('%s:%d:%s', gethostname(), Coroutine::getCid(), bin2hex(random_bytes(4)));
    }
}

4.2 Swoole 并发秒杀场景完整示例

<?php

useSwoole\Coroutine;
useSwoole\Coroutine\WaitGroup;
useApp\Lock\SwooleDistributedLock;

// 模拟 1000 个用户同时抢购（库存只有 100）
$redis = newRedis();
$redis->connect('127.0.0.1', 6379);
$redis->set('stock:flash:888', 100); // 初始库存 100

$lock = newSwooleDistributedLock($redis, ttl: 3, timeout: 2.0);
$successCount = 0;
$failCount = 0;

Coroutine\run(function () use ($redis, $lock, &$successCount, &$failCount) {
    $wg = newWaitGroup();

for ($i = 1; $i <= 1000; $i++) {
$wg->add();
go(function () use ($redis, $lock, &$successCount, &$failCount, $wg, $i) {
defer(fn() => $wg->done());

$resource = 'flash:888';

if (!$lock->acquire($resource)) {
$failCount++;  // 等锁超时
return;
            }

try {
$stock = (int) $redis->get('stock:flash:888');

if ($stock > 0) {
$redis->decr('stock:flash:888');
$successCount++;
// echo "用户 {$i} 抢购成功，剩余库存：" . ($stock - 1) . PHP_EOL;
                } else {
$failCount++;  // 库存不足
                }
            } finally {
$lock->release($resource);
            }
        });
    }

$wg->wait(10.0); // 最多等10秒
});

echo"=== 秒杀结果 ===" . PHP_EOL;
echo"成功购买：{$successCount} 件" . PHP_EOL;  // 应该输出 100
echo"失败次数：{$failCount} 次" . PHP_EOL;     // 应该输出 900
echo"最终库存：" . $redis->get('stock:flash:888') . PHP_EOL; // 应该输出 0

// 期望输出：
// === 秒杀结果 ===
// 成功购买：100 件  ✅ 精准等于初始库存
// 失败次数：900 次
// 最终库存：0

五、性能对比：各种方案的真实数据

用 wrk 对三种方案进行压测（1000 并发，持续 30s，初始库存 10000）：

结论：

• Swoole 协程锁的性能比 MySQL 行锁高 15 倍
• 比 PHP-FPM Redis 锁高 89%
• 性能逼近 Java Redisson（差距不到 15%）

六、生产环境五大坑，踩过的都懂

坑 1：锁 TTL 设太短，业务没跑完锁就过期了

// ❌ 问题：数据库操作需要 3 秒，但 TTL 只设了 1 秒
$lock = new Redis DistributedLock($redis, ttl: 1);

// ✅ 方案一：TTL 设为业务时间的 3-5 倍
$lock = newRedis DistributedLock($redis, ttl: 10);

// ✅ 方案二：开启自动续期（看门狗机制）
// 用一个独立协程/进程，每隔 TTL/3 时间续期一次
$timer = Swoole\Timer::tick(3000, function () use ($lock, $resource) {
    $lock->renew($resource); // 续期
});
// 业务完成后取消定时器
Swoole\Timer::clear($timer);

坑 2：Redis 主从切换时，锁丢失

场景：进程 A 在 master 加了锁，master 还没同步到 slave 就宕机
slave 升级为 master 后，锁不见了，进程 B 也能加锁
→ 临界区被两个进程同时进入

解决方案：用 Redlock 算法（同时向 N 个独立 Redis 节点加锁，过半数成功才算加锁成功）。

// RedLock 简化版（需要 3 个独立 Redis 节点）
classRedLock
{
privatearray$instances; // 3 或 5 个 Redis 连接
privateint$quorum;      // 过半数，3节点 = 2

publicfunctionacquire(string$resource, int$ttl): array|false
{
$owner = bin2hex(random_bytes(16));
$startTime = microtime(true) * 1000;
$acquired = 0;

foreach ($this->instances as$redis) {
if ($this->lockInstance($redis, $resource, $owner, $ttl)) {
$acquired++;
            }
        }

// 实际有效时间 = TTL - 获取锁消耗的时间 - 时钟漂移
$elapsedTime = microtime(true) * 1000 - $startTime;
$validityTime = $ttl - $elapsedTime - ($ttl * 0.01);

if ($acquired >= $this->quorum && $validityTime > 0) {
return ['owner' => $owner, 'validity' => $validityTime];
        }

// 没有获得过半数，释放已获得的锁
$this->unlockAll($resource, $owner);
returnfalse;
    }
}

注：Redlock 有一定争议（Martin Kleppmann vs Antirez 的著名论战），绝大多数业务场景用单节点 Redis 锁 + 合理的 TTL 就够了，不要过度工程化。

坑 3：锁粒度太粗，吞吐量上不去

// ❌ 粒度太粗：整个商品类目加一把锁
$lock->acquire('category:electronics');  // 所有电子产品都串行

// ✅ 粒度细化：精确到 SKU 级别
$lock->acquire("sku:{$skuId}");  // 只锁具体商品

// ✅ 更细：精确到用户+商品组合（防止同一用户重复下单）
$lock->acquire("order:{$userId}:{$skuId}");

坑 4：忘记 finally 释放锁，锁变"僵尸锁"

// ❌ 危险：业务抛异常后，锁没释放，只能等 TTL 自动过期
if ($lock->acquire($resource)) {
$result = doSomething(); // 如果这里抛异常...
$lock->release($resource); // 这行永远不会执行
}

// ✅ 正确：用 try/finally 或 withLock
$lock->withLock($resource, fn() => doSomething()); // 推荐

// 或者手动加 finally
if ($lock->acquire($resource)) {
try {
$result = doSomething();
    } finally {
$lock->release($resource); // 无论如何都会执行
    }
}

坑 5：可重入问题（同一个进程两次加同一把锁）

// 场景：外层加了锁，内层方法也尝试加同一把锁 → 死锁！
$lock->acquire('order:create');
// ... 业务逻辑
$this->deductStock($productId); // 内部也调用了 $lock->acquire('stock:xxx')
// 如果 stock:xxx 和 order:create 是同一个 key，就死锁了
$lock->release('order:create');

// 解决方案：每把锁的 key 要语义明确，避免嵌套同名锁
// 或者实现可重入锁（计数器）

七、完整的秒杀系统架构图

用户请求
    │
    ▼
[Nginx 限流] ← 第一道防线：每个 IP 限 QPS
    │
    ▼
[PHP-FPM / Swoole]
    │
    ├─ [本地缓存] ← 库存预减，减少 Redis 压力
    │       │
    │       ▼ （本地库存为0，直接返回失败）
    │
    ├─ [分布式锁] ← 第二道防线：互斥访问
    │   Redis SET NX EX
    │       │
    │       ▼ （加锁失败，返回"系统繁忙"）
    │
    ├─ [Redis 原子扣减] ← 第三道防线：原子操作
    │   DECRBY stock:xxx 1
    │       │
    │       ├─ 返回 >= 0：扣减成功，写消息队列
    │       └─ 返回 < 0：库存不足，INCRBY 回滚
    │
    ▼
[消息队列（RabbitMQ/Kafka）]
    │
    ▼
[异步消费：创建订单、扣减 DB 库存、发送通知]