IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

PHP 分布式事务TCC模式

wen PHP项目 2

本文目录导读：

PHP 分布式事务TCC模式

  1. TCC模式概述
  2. 核心架构设计
  3. 最佳实践和注意事项

我来详细讲解PHP中分布式事务TCC模式的实现。

TCC模式概述

TCC（Try-Confirm-Cancel）是一种补偿性分布式事务解决方案,包含三个阶段：

Try：预留业务资源
2. Confirm：确认执行业务
3. Cancel：取消预留资源

核心架构设计

基础接口定义

<?php
// TCC事务参与者接口
interface TccParticipant {
    // 预留资源
    public function try($transactionId, $params);
    // 确认执行
    public function confirm($transactionId);
    // 取消操作
    public function cancel($transactionId);
}
// 事务上下文
class TransactionContext {
    private $transactionId;
    private $status; // TRYING, CONFIRMING, CANCELLING
    private $participants = [];
    private $params = [];
    public function __construct($transactionId) {
        $this->transactionId = $transactionId;
        $this->status = 'TRYING';
    }
    public function addParticipant(TccParticipant $participant, $params) {
        $this->participants[] = [
            'participant' => $participant,
            'params' => $params
        ];
    }
    // getters/setters...
}

事务管理器

<?php
class TccTransactionManager {
    private $cache;
    private $logger;
    private $maxRetries = 3;
    public function __construct($cache, $logger) {
        $this->cache = $cache;
        $this->logger = $logger;
    }
    /**
     * 执行TCC事务
     */
    public function execute(callable $transactionFunc) {
        $transactionId = $this->generateTransactionId();
        $context = new TransactionContext($transactionId);
        try {
            // 阶段一：Try
            $transactionFunc($context);
            $this->executeTryPhase($context);
            // 阶段二：Confirm
            return $this->executeConfirmPhase($context);
        } catch (Exception $e) {
            // 阶段三：Cancel
            $this->executeCancelPhase($context);
            throw $e;
        }
    }
    /**
     * 执行Try阶段
     */
    private function executeTryPhase(TransactionContext $context) {
        $this->logger->info("开始Try阶段", ['transactionId' => $context->getTransactionId()]);
        $successfulParticipants = [];
        foreach ($context->getParticipants() as $index => $participantInfo) {
            try {
                $participant = $participantInfo['participant'];
                $params = $participantInfo['params'];
                // 执行Try
                $result = $participant->try(
                    $context->getTransactionId(), 
                    $params
                );
                $successfulParticipants[] = $index;
                // 保存执行状态
                $this->saveTransactionState($context, $participant, 'TRY_SUCCESS');
            } catch (Exception $e) {
                $this->logger->error("Try阶段失败", [
                    'transactionId' => $context->getTransactionId(),
                    'participant' => get_class($participant),
                    'error' => $e->getMessage()
                ]);
                // 回滚已成功的Try操作
                foreach ($successfulParticipants as $successIndex) {
                    $this->executeCancelForParticipant($context, $successIndex);
                }
                throw new TccException("Try阶段失败", 0, $e);
            }
        }
        $context->setStatus('TRY_SUCCESS');
        $this->saveContext($context);
    }
    /**
     * 执行Confirm阶段
     */
    private function executeConfirmPhase(TransactionContext $context) {
        $this->logger->info("开始Confirm阶段", ['transactionId' => $context->getTransactionId()]);
        try {
            $results = [];
            foreach ($context->getParticipants() as $index => $participantInfo) {
                $participant = $participantInfo['participant'];
                $retryCount = 0;
                while ($retryCount < $this->maxRetries) {
                    try {
                        $result = $participant->confirm($context->getTransactionId());
                        $results[] = $result;
                        break;
                    } catch (Exception $e) {
                        $retryCount++;
                        if ($retryCount >= $this->maxRetries) {
                            $this->logger->error("Confirm阶段重试失败", [
                                'transactionId' => $context->getTransactionId(),
                                'participant' => get_class($participant),
                                'retryCount' => $retryCount
                            ]);
                            // 记录异常，需要人工介入
                            $this->recordManualIntervention($context, $participant, $e);
                        }
                        sleep(1);
                    }
                }
            }
            $context->setStatus('CONFIRM_SUCCESS');
            $this->destroyContext($context);
            return $results;
        } catch (Exception $e) {
            $this->logger->critical("Confirm阶段异常，需要人工处理", [
                'transactionId' => $context->getTransactionId(),
                'error' => $e->getMessage()
            ]);
            throw $e;
        }
    }
    /**
     * 执行Cancel阶段
     */
    private function executeCancelPhase(TransactionContext $context) {
        $this->logger->info("执行Cancel阶段", ['transactionId' => $context->getTransactionId()]);
        $participants = $context->getParticipants();
        foreach ($participants as $index => $participantInfo) {
            $this->executeCancelForParticipant($context, $index);
        }
        $context->setStatus('CANCEL_SUCCESS');
        $this->saveContext($context);
    }
    private function executeCancelForParticipant($context, $index) {
        $participantInfo = $context->getParticipants()[$index];
        $participant = $participantInfo['participant'];
        try {
            $participant->cancel($context->getTransactionId());
            $this->logger->info("Cancel成功", [
                'transactionId' => $context->getTransactionId(),
                'participant' => get_class($participant)
            ]);
        } catch (Exception $e) {
            $this->logger->error("Cancel失败", [
                'transactionId' => $context->getTransactionId(),
                'participant' => get_class($participant),
                'error' => $e->getMessage()
            ]);
            // 保存状态以便后续补偿
            $this->saveCancellationFailure($context, $participant);
        }
    }
    // 生成事务ID
    private function generateTransactionId() {
        return uniqid('tcc_', true) . bin2hex(random_bytes(8));
    }
    // 保存事务状态到缓存
    private function saveContext(TransactionContext $context) {
        $key = "tcc:transaction:{$context->getTransactionId()}";
        $this->cache->set($key, serialize($context), 3600);
    }
    private function destroyContext(TransactionContext $context) {
        $key = "tcc:transaction:{$context->getTransactionId()}";
        $this->cache->delete($key);
    }
    // 获取事务上下文
    public function getContext($transactionId) {
        $key = "tcc:transaction:{$transactionId}";
        $data = $this->cache->get($key);
        return $data ? unserialize($data) : null;
    }
    // 保存状态
    private function saveTransactionState($context, $participant, $status) {
        // 保存到数据库或日志
        $state = [
            'transaction_id' => $context->getTransactionId(),
            'participant' => get_class($participant),
            'status' => $status,
            'time' => date('Y-m-d H:i:s')
        ];
        $this->logger->info("保存事务状态", $state);
    }
    private function recordManualIntervention($context, $participant, $e) {
        $this->logger->critical("需要人工介入的异常事务", [
            'transaction_id' => $context->getTransactionId(),
            'participant' => get_class($participant),
            'status' => 'MANUAL_INTERVENTION_REQUIRED',
            'exception' => $e->getMessage()
        ]);
    }
    private function saveCancellationFailure($context, $participant) {
        $this->logger->error("取消操作失败，等待后续处理", [
            'transaction_id' => $context->getTransactionId(),
            'participant' => get_class($participant),
            'status' => 'CANCEL_FAILED'
        ]);
    }
}
// 自定义异常
class TccException extends Exception {}

具体参与者实现示例

<?php
// 账户服务参与者
class AccountServiceParticipant implements TccParticipant {
    private $db;
    public function __construct($db) {
        $this->db = $db;
    }
    /**
     * Try阶段：冻结资金
     */
    public function try($transactionId, $params) {
        $accountId = $params['account_id'];
        $amount = $params['amount'];
        $this->db->beginTransaction();
        try {
            // 检查余额是否充足
            $balance = $this->db->query(
                "SELECT balance FROM accounts WHERE id = ? FOR UPDATE",
                [$accountId]
            );
            if ($balance < $amount) {
                throw new \Exception("余额不足");
            }
            // 创建冻结记录
            $this->db->execute(
                "INSERT INTO account_frozen (transaction_id, account_id, amount, status, created_at) 
                 VALUES (?, ?, ?, 'FROZEN', NOW())",
                [$transactionId, $accountId, $amount]
            );
            // 更新可用余额
            $this->db->execute(
                "UPDATE accounts SET frozen_amount = frozen_amount + ? WHERE id = ?",
                [$amount, $accountId]
            );
            $this->db->commit();
            return true;
        } catch (Exception $e) {
            $this->db->rollback();
            throw $e;
        }
    }
    /**
     * Confirm阶段：完成资金扣减
     */
    public function confirm($transactionId) {
        $this->db->beginTransaction();
        try {
            // 获取冻结记录
            $frozen = $this->db->query(
                "SELECT * FROM account_frozen WHERE transaction_id = ? AND status = 'FROZEN'",
                [$transactionId]
            );
            if (!$frozen) {
                throw new \Exception("无效的冻结记录");
            }
            // 更新账户余额
            $this->db->execute(
                "UPDATE accounts SET 
                    balance = balance - ?,
                    frozen_amount = frozen_amount - ?
                 WHERE id = ?",
                [$frozen['amount'], $frozen['amount'], $frozen['account_id']]
            );
            // 更新冻结记录状态
            $this->db->execute(
                "UPDATE account_frozen SET status = 'CONFIRMED', confirmed_at = NOW() 
                 WHERE transaction_id = ?",
                [$transactionId]
            );
            $this->db->commit();
            return true;
        } catch (Exception $e) {
            $this->db->rollback();
            throw $e;
        }
    }
    /**
     * Cancel阶段：释放冻结资金
     */
    public function cancel($transactionId) {
        $this->db->beginTransaction();
        try {
            // 获取冻结记录
            $frozen = $this->db->query(
                "SELECT * FROM account_frozen WHERE transaction_id = ? AND status = 'FROZEN'",
                [$transactionId]
            );
            if ($frozen) {
                // 解冻资金
                $this->db->execute(
                    "UPDATE accounts SET frozen_amount = frozen_amount - ? WHERE id = ?",
                    [$frozen['amount'], $frozen['account_id']]
                );
                // 更新冻结记录状态
                $this->db->execute(
                    "UPDATE account_frozen SET status = 'CANCELLED', cancelled_at = NOW() 
                     WHERE transaction_id = ?",
                    [$transactionId]
                );
            }
            $this->db->commit();
            return true;
        } catch (Exception $e) {
            $this->db->rollback();
            throw $e;
        }
    }
}

库存服务参与者

<?php
// 库存服务参与者
class InventoryServiceParticipant implements TccParticipant {
    private $redis;
    private $db;
    public function __construct($redis, $db) {
        $this->redis = $redis;
        $this->db = $db;
    }
    /**
     * Try阶段：预扣库存
     */
    public function try($transactionId, $params) {
        $productId = $params['product_id'];
        $quantity = $params['quantity'];
        $this->db->beginTransaction();
        try {
            // 检查库存
            $inventory = $this->db->query(
                "SELECT * FROM inventory WHERE product_id = ? FOR UPDATE",
                [$productId]
            );
            if ($inventory['quantity'] < $quantity) {
                throw new \Exception("库存不足");
            }
            // 创建预扣记录
            $this->db->execute(
                "INSERT INTO inventory_reservation (transaction_id, product_id, quantity, status, created_at) 
                 VALUES (?, ?, ?, 'RESERVED', NOW())",
                [$transactionId, $productId, $quantity]
            );
            $this->db->commit();
            return true;
        } catch (Exception $e) {
            $this->db->rollback();
            throw $e;
        }
    }
    // confirm和cancel方法类似...
}

使用示例

<?php
// 使用TCC事务示例
$tccManager = new TccTransactionManager($cache, $logger);
try {
    $results = $tccManager->execute(function($context) use ($userId, $productId, $amount, $accountService, $inventoryService) {
        // 添加账户扣款参与者
        $context->addParticipant($accountService, [
            'account_id' => $userId,
            'amount' => $amount
        ]);
        // 添加库存扣减参与者
        $context->addParticipant($inventoryService, [
            'product_id' => $productId,
            'quantity' => 1
        ]);
        // 可以添加更多参与者
    });
    echo "事务执行成功";
} catch (TccException $e) {
    echo "事务执行失败：" . $e->getMessage();
}

异常补偿机制

<?php
// 补偿任务管理器
class CompensationManager {
    private $db;
    private $logger;
    /**
     * 扫描并处理失败的Cancel操作
     */
    public function processFailedCancellations() {
        $query = "SELECT * FROM transaction_compensation 
                  WHERE status = 'PENDING' OR status = 'RETRYING'";
        $records = $this->db->query($query);
        foreach ($records as $record) {
            $this->retryCompensation($record);
        }
    }
    /**
     * 重试补偿操作
     */
    private function retryCompensation($record) {
        $maxAttempts = 3;
        $attempt = 0;
        while ($attempt < $maxAttempts) {
            try {
                // 重新执行Cancel
                $participant = $this->getParticipant($record['participant_class']);
                $participant->cancel($record['transaction_id']);
                // 更新状态
                $this->db->execute(
                    "UPDATE transaction_compensation SET status = 'COMPLETED', 
                     completed_at = NOW() WHERE id = ?",
                    [$record['id']]
                );
                break;
            } catch (Exception $e) {
                $attempt++;
                sleep(2 * $attempt); // 递增等待时间
                if ($attempt >= $maxAttempts) {
                    $this->logger->error("补偿操作最终失败", [
                        'recordId' => $record['id'],
                        'transactionId' => $record['transaction_id']
                    ]);
                    // 标记为失败，需要人工处理
                    $this->db->execute(
                        "UPDATE transaction_compensation SET status = 'NEED_MANUAL', 
                         error_message = ? WHERE id = ?",
                        [$e->getMessage(), $record['id']]
                    );
                }
            }
        }
    }
}

数据库表结构

-- 冻结记录表
CREATE TABLE account_frozen (
    id INT AUTO_INCREMENT PRIMARY KEY,
    transaction_id VARCHAR(64) NOT NULL,
    account_id INT NOT NULL,
    amount DECIMAL(10,2) NOT NULL,
    status ENUM('FROZEN', 'CONFIRMED', 'CANCELLED') DEFAULT 'FROZEN',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    confirmed_at TIMESTAMP NULL,
    cancelled_at TIMESTAMP NULL,
    INDEX idx_transaction_id (transaction_id),
    INDEX idx_account_id (account_id)
);
-- 库存预留表
CREATE TABLE inventory_reservation (
    id INT AUTO_INCREMENT PRIMARY KEY,
    transaction_id VARCHAR(64) NOT NULL,
    product_id INT NOT NULL,
    quantity INT NOT NULL,
    status ENUM('RESERVED', 'CONFIRMED', 'CANCELLED') DEFAULT 'RESERVED',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_transaction_id (transaction_id)
);
-- 事务状态表
CREATE TABLE tcc_transaction (
    id INT AUTO_INCREMENT PRIMARY KEY,
    transaction_id VARCHAR(64) NOT NULL,
    status VARCHAR(20) NOT NULL,
    participants TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    UNIQUE KEY uk_transaction_id (transaction_id)
);
-- 需要人工干预的事务记录表
CREATE TABLE tcc_manual_intervention (
    id INT AUTO_INCREMENT PRIMARY KEY,
    transaction_id VARCHAR(64) NOT NULL,
    participant_class VARCHAR(200),
    error_message TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    status ENUM('PENDING', 'PROCESSED') DEFAULT 'PENDING'
);

最佳实践和注意事项

幂等性设计

// 幂等性检查示例
class IdempotencyHandler {
    private $redis;
    public function check($transactionId, $operation) {
        $key = "tcc:idempotency:{$transactionId}:{$operation}";
        // 使用Redis的原子操作
        $result = $this->redis->setnx($key, 1);
        if ($result) {
            // 设置过期时间（例如10分钟）
            $this->redis->expire($key, 600);
            return true; // 首次操作
        }
        return false; // 重复操作
    }
}

超时处理

// 设置合理的超时时间
class TimeoutHandler {
    const TRY_TIMEOUT = 5000;      // 5秒
    const CONFIRM_TIMEOUT = 10000; // 10秒
    const CANCEL_TIMEOUT = 10000;  // 10秒
    public function setTimeout($operation, $timeout) {
        // 设置HTTP客户端超时
        $this->client->setTimeout($timeout);
        // 设置数据库操作超时
        $this->db->setQueryTimeout($timeout);
    }
}

性能优化建议

  • 使用连接池管理数据库连接
  • 批量处理操作命令
  • 使用异步消息队列处理非核心业务
  • 合理设置缓存过期时间
// 连接池示例
class ConnectionPool {
    private $pool = [];
    private $maxSize = 10;
    public function getConnection() {
        // 从池中获取连接或创建新连接
    }
    public function releaseConnection($connection) {
        // 归还连接
    }
}

TCC模式通过业务拆分配置，提供了可靠的分布式事务解决方案，特别适合对数据一致性要求较高的金融、电商等场景，但要注意其性能开销相对较高,需要根据业务场景合理选择。

上一篇PHP 最大努力通知方案

下一篇PHP 事务消息怎么实现

相关文章

抱歉，评论功能暂时关闭!