Laravel多控制台命令并发Upsert序列化死锁问题排查与解决

核心原因分析 #

死锁发生在多Worker并行执行upsert时，MySQL的INSERT ... ON DUPLICATE KEY UPDATE会对冲突行加排他锁。当多个Worker的操作涉及不同行但锁获取顺序不一致时，就会出现循环等待，触发1213错误。即便你给worker字段加了唯一索引，若upsert的唯一键包含其他字段，或者批量操作的锁范围交叉，仍会引发死锁。

具体解决方案 #

1. 调整Upsert的唯一键与索引策略 #

• 确保upsert使用的唯一键是联合唯一索引，且包含worker字段，让每个Worker的操作范围被严格隔离。比如如果聚合表按worker+stat_date+type做唯一标识，创建对应联合唯一索引：

  // 迁移文件示例
  $table->unique(['worker', 'stat_date', 'type']);
  

• 避免单独给worker加唯一索引（除非单Worker对应唯一行），而是让唯一键明确区分每个Worker的操作单元，减少跨Worker的锁竞争。

2. 拆分批量Upsert为更小批次 #

大批次upsert会持有锁的时间更长，且更容易触发范围锁。将每批处理的数据从数千条拆分为100-200条的小批次，降低锁冲突概率：

// 核心处理代码调整
$streamData = Redis::xReadGroup('stat_consumer', $worker, ['stat_stream' => '>'], 200); // 拉取200条
foreach (array_chunk($streamData['stat_stream'], 50) as $chunk) { // 再拆分为50条小批次
    $records = collect($chunk)->map(function ($item) {
        return $this->formatRecord($item[1]);
    })->toArray();

    DB::table('stat_aggregates')->upsert(
        $records,
        ['worker', 'stat_date', 'type'], // 联合唯一键
        ['count', 'sum_value'] // 更新字段
    );
}

3. 显式控制事务与锁顺序 #

• 对每个小批次的upsert单独开启事务，避免长时间持有锁：

foreach ($chunks as $chunk) {
    DB::transaction(function () use ($chunk) {
        DB::table('stat_aggregates')->upsert(
            $chunk,
            ['worker', 'stat_date', 'type'],
            ['count', 'sum_value']
        );
    });
}

• 强制Worker按固定顺序处理数据，比如对每个批次内的记录按唯一键排序后再执行upsert，确保所有Worker获取锁的顺序一致，避免循环等待：

$records = collect($chunk)->map(function ($item) {
    return $this->formatRecord($item[1]);
})->sortBy(function ($record) {
    return $record['worker'] . $record['stat_date'] . $record['type']; // 按唯一键排序
})->toArray();

4. 使用MySQL的低隔离级别 #

将事务隔离级别调整为READ COMMITTED，这会减少InnoDB的锁范围，避免不必要的间隙锁：

// 在命令类的handle方法开头设置
DB::statement("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED");

或者在config/database.php的mysql配置中默认设置：

'mysql' => [
    // ...其他配置
    'isolation_level' => 'READ COMMITTED',
],

5. 死锁重试机制 #

即便做了以上优化，极端情况下仍可能出现死锁，因此添加重试逻辑是必要的：

function upsertWithRetry($records, $uniqueBy, $update) {
    $attempts = 3;
    while ($attempts-- > 0) {
        try {
            return DB::table('stat_aggregates')->upsert($records, $uniqueBy, $update);
        } catch (\Illuminate\Database\QueryException $e) {
            if ($e->getCode() === '40001' || $e->getCode() === '1213') { // 死锁错误码
                usleep(100000); // 等待100ms后重试
                continue;
            }
            throw $e;
        }
    }
    throw new \Exception("Upsert failed after 3 attempts due to deadlocks");
}

// 使用示例
upsertWithRetry($records, ['worker', 'stat_date', 'type'], ['count', 'sum_value']);

验证与监控 #

• 开启MySQL的死锁日志（innodb_print_all_deadlocks=1），持续观察死锁的具体场景，验证优化效果。
• 监控Worker的执行时间与锁等待情况，调整批次大小和重试策略。

内容的提问来源于stack exchange，提问作者Ryan H