本文目录导读:

在PHP API架构中保证最终一致性,通常涉及以下核心策略和实践:
异步消息队列(最常用)
实现方式
// 使用Redis作为消息队列
class AsyncEventPublisher {
private $redis;
public function publish($event, $data) {
$message = json_encode([
'event' => $event,
'data' => $data,
'timestamp' => time()
]);
$this->redis->lPush('event_queue', $message);
}
}
// 消费者进程
class EventConsumer {
public function process() {
while ($message = $this->redis->brPop('event_queue', 5)) {
$event = json_decode($message, true);
$this->handleEvent($event);
}
}
}
事件溯源(Event Sourcing)
// 记录所有状态变更事件
class OrderService {
private $eventStore;
public function createOrder($data) {
// 1. 保存订单(主数据)
$orderId = $this->db->insert('orders', $data);
// 2. 记录事件
$this->eventStore->append('order.created', [
'order_id' => $orderId,
'data' => $data,
'version' => 1
]);
// 3. 发布异步事件
EventBus::publish('order.created', ['order_id' => $orderId]);
return $orderId;
}
}
补偿事务(Saga模式)
class SagaOrchestrator {
private $steps = [];
private $compensationSteps = [];
public function execute() {
$completedSteps = [];
try {
foreach ($this->steps as $index => $step) {
$result = $step->execute();
$completedSteps[] = $index;
}
return true;
} catch (\Exception $e) {
// 回滚已完成的步骤
foreach (array_reverse($completedSteps) as $index) {
$this->compensationSteps[$index]->compensate();
}
throw $e;
}
}
}
读修复(Read Repair)
class EventuallyConsistentReader {
private $cache;
private $db;
public function getWithRepair($key) {
// 尝试从缓存读取
$data = $this->cache->get($key);
if (!$data) {
// 缓存未命中,从数据库读取
$dbData = $this->db->find($key);
if ($dbData) {
// 修复缓存(异步)
$this->cache->setAsync($key, $dbData);
return $dbData;
}
// 从其他副本读取并对比
$replicas = $this->getFromReplicas($key);
$consistentData = $this->resolveConflict($replicas);
// 修复不一致的副本
$this->repairReplicas($key, $consistentData);
return $consistentData;
}
return $data;
}
}
版本向量/时钟
class VersionVector {
private $versions = [];
public function increment($nodeId) {
$this->versions[$nodeId] = ($this->versions[$nodeId] ?? 0) + 1;
}
public function merge(VersionVector $other) {
foreach ($other->versions as $nodeId => $version) {
$this->versions[$nodeId] = max(
$this->versions[$nodeId] ?? 0,
$version
);
}
}
public function compare(VersionVector $other) {
// 返回 -1: 小于, 0: 冲突, 1: 大于
$hasNewer = false;
$hasOlder = false;
foreach ($this->versions as $nodeId => $version) {
$otherVersion = $other->versions[$nodeId] ?? 0;
if ($version > $otherVersion) $hasNewer = true;
if ($version < $otherVersion) $hasOlder = true;
}
if ($hasNewer && !$hasOlder) return 1;
if (!$hasNewer && $hasOlder) return -1;
if ($hasNewer && $hasOlder) return 0;
return 0; // 相等
}
}
幂等性保证
class IdempotentService {
private $processedRequests;
public function processRequest($requestId, $callback) {
// 检查是否已处理
if ($this->processedRequests->exists($requestId)) {
return $this->processedRequests->get($requestId);
}
// 执行操作
$result = $callback();
// 记录已处理
$this->processedRequests->set($requestId, $result, 3600); // 1小时过期
return $result;
}
}
最终一致性检查器
class ConsistencyChecker {
private $db;
private $es;
public function checkAndRepair() {
// 定期比较数据库和搜索引擎
$dbOrders = $this->db->query('SELECT id, updated_at FROM orders WHERE updated_at > ?', [time() - 3600]);
foreach ($dbOrders as $order) {
$esOrder = $this->es->get('orders', $order['id']);
if (!$esOrder || $esOrder['updated_at'] < $order['updated_at']) {
// 更新搜索引擎
$fullOrder = $this->db->find($order['id']);
$this->es->index('orders', $fullOrder);
}
}
}
}
最佳实践
选择合适的一致性模型
- 强一致性:金融交易、库存扣减
- 最终一致性:搜索索引、缓存、统计
监控告警
class ConsistencyMonitor {
public function measureConsistencyLag() {
$dbTime = $this->db->query('SELECT MAX(updated_at) FROM orders');
$cacheTime = $this->cache->get('max_order_time');
$lag = $dbTime - $cacheTime;
if ($lag > 300) { // 超过5分钟
Alert::warning("Consistency lag detected: {$lag}s");
}
}
}
渐进式保证
- 先保证最终一致性
- 再逐步优化延迟
- 最后考虑强一致性场景
最终一致性的核心是接受短时间的不一致,但保证最终会一致,关键在于设计好补偿机制和冲突解决策略。