怎样在PHP项目中实现分布式事务？

本文目录导读：

1. 基于消息队列的最终一致性方案（推荐）
2. TCC（Try-Confirm-Cancel）模式
3. Saga模式
4. 使用XA协议（两阶段提交）
5. 使用成熟的分布式事务框架
6. 选择建议
7. 实际项目最佳实践

在PHP项目中实现分布式事务是一个复杂但常见的问题,由于PHP本身是单线程、无状态的语言，通常需要借助外部工具或架构模式来实现分布式事务。

以下是几种在PHP中实现分布式事务的主流方案：

基于消息队列的最终一致性方案（推荐）

这是最常用且性能最好的方案,适用于对实时一致性要求不高的场景。

核心思想：本地消息表 + 消息队列

// 1. 本地业务操作 + 消息记录在同一个数据库事务中
try {
    $pdo->beginTransaction();
    // 业务操作 - 扣减库存
    $pdo->exec("UPDATE inventory SET stock = stock - 1 WHERE product_id = 1");
    // 记录待发送消息
    $pdo->exec("INSERT INTO message_queue (status, content) VALUES (0, '{"order_id": 123}')");
    $pdo->commit();
    // 2. 异步发送消息到MQ
    $mq->send('order_topic', json_encode(['order_id' => 123]));
} catch (Exception $e) {
    $pdo->rollBack();
}
// 3. 消息消费端处理（另一个服务）
class OrderConsumer {
    public function handle($message) {
        // 消费消息，更新订单状态
        $orderService->updateStatus($message['order_id'], 'paid');
        // 如果失败，消息会重试
    }
}

TCC（Try-Confirm-Cancel）模式

适用于对一致性要求较高的场景。

// TCC接口定义
interface TCCService {
    public function try($params); // 预留资源
    public function confirm($params); // 确认执行
    public function cancel($params); // 取消回滚
}
// TCC协调器
class TCCCoordinator {
    private $participants = [];
    public function executeTransaction() {
        // 1. Try阶段 - 尝试所有参与者
        foreach ($this->participants as $p) {
            try {
                $p->try($params);
                $tried[] = $p;
            } catch (Exception $e) {
                // 2. 如果失败，取消所有已Try的参与者
                foreach ($tried as $t) {
                    $t->cancel($params); // 可能也需要失败重试
                }
                throw $e;
            }
        }
        // 3. Confirm阶段 - 如果所有Try成功
        foreach ($tried as $p) {
            try {
                $p->confirm($params);
            } catch (Exception $e) {
                // 需要补偿机制，可能还需要重试
                throw $e;
            }
        }
    }
}
// 具体实现
class PaymentService implements TCCService {
    public function try($params) {
        // 冻结用户资金
        $this->freezeBalance($params['user_id'], $params['amount']);
    }
    public function confirm($params) {
        // 扣减冻结资金
        $this->deductBalance($params['user_id'], $params['amount']);
    }
    public function cancel($params) {
        // 解冻用户资金
        $this->unfreezeBalance($params['user_id'], $params['amount']);
    }
}

Saga模式

通过一系列的本地事务和补偿操作来实现。

class Saga {
    private $steps = [];
    private $compensations = [];
    private $executedSteps = [];
    public function addStep($action, $compensation) {
        $this->steps[] = $action;
        $this->compensations[] = $compensation;
    }
    public function execute() {
        try {
            foreach ($this->steps as $i => $step) {
                $step();
                $this->executedSteps[] = $i;
            }
        } catch (Exception $e) {
            // 回滚已执行的操作
            foreach (array_reverse($this->executedSteps) as $i) {
                try {
                    $this->compensations[$i]();
                } catch (Exception $compException) {
                    // 补偿失败需要记录并人工处理
                    Log::error("Saga compensation failed: " . $compException->getMessage());
                }
            }
            throw $e;
        }
    }
}
// 使用
$saga = new Saga();
$saga->addStep(
    function() { /* 扣减库存 */ },
    function() { /* 回滚库存 */ }
);
$saga->addStep(
    function() { /* 创建订单 */ },
    function() { /* 取消订单 */ }
);
$saga->addStep(
    function() { /* 发送积分 */ },
    function() { /* 回滚积分 */ }
);
$saga->execute();

使用XA协议（两阶段提交）

需要数据库和消息队列支持XA协议。

// 使用PDO的XA支持
try {
    // 开始全局事务
    $pdo->exec("XA START 'global_transaction_1'");
    // 操作数据库A
    $pdoA->exec("XA END 'global_transaction_1'");
    $pdoA->exec("XA PREPARE 'global_transaction_1'");
    // 操作数据库B（如果支持）
    $pdoB->exec("XA END 'global_transaction_1'");
    $pdoB->exec("XA PREPARE 'global_transaction_1'");
    // 提交
    $pdoA->exec("XA COMMIT 'global_transaction_1'");
    $pdoB->exec("XA COMMIT 'global_transaction_1'");
} catch (Exception $e) {
    // 回滚
    $pdoA->exec("XA ROLLBACK 'global_transaction_1'");
    $pdoB->exec("XA ROLLBACK 'global_transaction_1'");
}

使用成熟的分布式事务框架

Seata (推荐)

// Seata-php客户端
$seata = new SeataClient();
$seata->beginTransaction();
try {
    // Service A - 扣减库存
    $inventoryService->deduct($productId, 1);
    // Service B - 创建订单
    $orderService->create($userId, $productId, 1);
    $seata->commit();
} catch (Exception $e) {
    $seata->rollback();
}

选择建议

实际项目最佳实践

1. 优先选择最终一致性方案

   • 大部分业务场景可以接受短时间不一致
   • 性能好,易于维护

2. 合理使用异步和消息队列

   RabbitMQ, Kafka, RocketMQ都支持事务消息

3. 做好补偿机制

   • 实现幂等性
   • 记录操作日志
   • 提供人工干预接口

4. 监控和告警

   • 监控事务成功率
   • 对失败的事务进行告警

5. 避免过度设计

   • 如果服务间调用量不大,可以考虑使用2PC+重试
   • 如果业务对一致性要求不高,可以考虑异步队列

最推荐的做法是：优先使用消息队列实现最终一致性，只有在资金、库存等对实时一致性要求极高的场景才使用TCC或Saga模式。