데이터베이스 샤딩 확장을 구현하는 PHP 방법

// 主库配置信息
$masterConfig = [
    'dsn'      => 'mysql:host=127.0.0.1;port=3306;dbname=test',
    'username' => 'root',
    'password' => 'root',
];

// 分片库列表
$shardConfigList = [
    [
        'dsn'      => 'mysql:host=127.0.0.1;port=3306;dbname=test_shard_0',
        'username' => 'root',
        'password' => 'root',
    ],
    [
        'dsn'      => 'mysql:host=127.0.0.1;port=3306;dbname=test_shard_1',
        'username' => 'root',
        'password' => 'root',
    ],
];

// 创建PDO连接
$pdo = new PDO($masterConfig['dsn'], $masterConfig['username'], $masterConfig['password'], [
    PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
    PDO::ATTR_EMULATE_PREPARES => false,
]);

// 获取分片信息
$sql = 'SELECT * FROM `shard_info` WHERE shard_id = ?';
$stmt = $pdo->prepare($sql);
$stmt->execute([$shardId]);
$info = $stmt->fetch(PDO::FETCH_ASSOC);
if (!$info) {
    throw new RuntimeException('Shard not found');
}

// 创建分片PDO连接
$pdoShard = new PDO($info['dsn'], $info['username'], $info['password'], [
    PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
    PDO::ATTR_EMULATE_PREPARES => false,
]);

// 执行查询
$sql = 'SELECT * FROM `table` WHERE `key` = ?';
$stmt = $pdoShard->prepare($sql);
$stmt->execute([$key]);
$rows = $stmt->fetchAll(PDO::FETCH_ASSOC);

// 开始分布式事务
$pdo->beginTransaction();

try {
    // 准备分片事务
    $xid = uniqid();
    $prepares = [];

    foreach ($shardConfigList as $shardConfig) {
        $pdoShard = new PDO($shardConfig['dsn'], $shardConfig['username'], $shardConfig['password'], [
            PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
            PDO::ATTR_EMULATE_PREPARES => false,
        ]);

        $pdoShard->beginTransaction();
        $stmt = $pdoShard->prepare('INSERT INTO `table` (`key`, `value`) VALUES (?, ?)');
        $stmt->execute([$key, $value]);
        $prepares[] = [$pdoShard, $xid];
    }

    // 提交分片事务
    foreach ($prepares as [$pdoShard, $xid]) {
        $stmt = $pdoShard->prepare('PREPARE TRANSACTION ?');
        $stmt->execute([$xid]);
    }

    foreach ($prepares as [$pdoShard, $xid]) {
        $stmt = $pdoShard->prepare('COMMIT PREPARED ?');
        $stmt->execute([$xid]);
    }

    // 提交整个事务
    $pdo->commit();
} catch (Exception $ex) {
    // 回滚分片事务
    foreach ($prepares as [$pdoShard, $xid]) {
        $stmt = $pdoShard->prepare('ROLLBACK PREPARED ?');
        $stmt->execute([$xid]);
    }

    // 回滚整个事务
    $pdo->rollback();
}