PHP 怎么线性一致性

wen PHP项目 2026-08-04 279

本文目录导读：

PHP 怎么线性一致性

使用 Redis 分布式锁（最常用）
使用 PostgreSQL 事务（最强保证）
使用原子操作（Redis Lua 脚本）
使用分布式系统（etcd/Consul）
性能优化建议

在 PHP 中实现线性一致性（Linearizability）通常不是语言层面的特性，而是需要通过并发控制和外部存储来保证，以下是几种常见方案：

使用 Redis 分布式锁（最常用）

<?php
class LinearizableCounter {
    private $redis;
    private $lockKey;
    public function __construct($redis) {
        $this->redis = $redis;
        $this->lockKey = 'lock:counter';
    }
    // 原子递增操作（线性一致）
    public function increment($userId) {
        // 获取分布式锁
        $lock = $this->acquireLock();
        try {
            // 读取当前值
            $current = $this->redis->get("counter:{$userId}") ?: 0;
            // 执行操作
            $newValue = $current + 1;
            $this->redis->set("counter:{$userId}", $newValue);
            return $newValue;
        } finally {
            // 释放锁
            $this->releaseLock($lock);
        }
    }
    private function acquireLock($timeout = 10) {
        $uniqueId = uniqid('lock:', true);
        $start = time();
        // 尝试获取锁
        while (time() - $start < $timeout) {
            if ($this->redis->set($this->lockKey, $uniqueId, ['NX', 'EX' => 30])) {
                return $uniqueId;
            }
            usleep(100000); // 100ms
        }
        throw new Exception("Failed to acquire lock");
    }
    private function releaseLock($lockId) {
        // 使用 Lua 脚本确保原子性
        $script = "
            if redis.call('get', KEYS[1]) == ARGV[1] then
                return redis.call('del', KEYS[1])
            end
            return 0
        ";
        $this->redis->eval($script, [$this->lockKey, $lockId], 1);
    }
}

使用 PostgreSQL 事务（最强保证）

<?php
class LinearizableAccount {
    private $pdo;
    public function __construct($pdo) {
        $this->pdo = $pdo;
    }
    // 转账操作（线性一致）
    public function transfer($fromAccount, $toAccount, $amount) {
        $this->pdo->beginTransaction();
        try {
            // 锁定源账户
            $stmt = $this->pdo->prepare(
                "SELECT balance FROM accounts 
                 WHERE id = ? FOR UPDATE"
            );
            $stmt->execute([$fromAccount]);
            $fromBalance = $stmt->fetchColumn();
            // 检查余额
            if ($fromBalance < $amount) {
                throw new Exception("Insufficient funds");
            }
            // 更新源账户
            $this->pdo->prepare(
                "UPDATE accounts SET balance = balance - ? WHERE id = ?"
            )->execute([$amount, $fromAccount]);
            // 更新目标账户
            $this->pdo->prepare(
                "UPDATE accounts SET balance = balance + ? WHERE id = ?"
            )->execute([$amount, $toAccount]);
            // 提交事务
            $this->pdo->commit();
        } catch (Exception $e) {
            $this->pdo->rollBack();
            throw $e;
        }
    }
}

使用原子操作（Redis Lua 脚本）

<?php
class AtomicOperations {
    private $redis;
    // 使用 Lua 脚本实现原子操作（线性一致）
    public function checkAndSet($key, $expectedValue, $newValue) {
        $script = <<<LUA
            local current = redis.call('GET', KEYS[1])
            if current == ARGV[1] then
                redis.call('SET', KEYS[1], ARGV[2])
                return 1
            end
            return 0
        LUA;
        return $this->redis->eval(
            $script, 
            [$key, $expectedValue, $newValue], 
            1
        );
    }
    // 乐观锁实现
    public function optimisticUpdate($key, $updateFunc) {
        $retryCount = 0;
        do {
            // 读取当前版本
            $current = $this->redis->get($key);
            $newValue = $updateFunc($current);
            // 使用 WATCH/MULTI/EXEC 实现乐观锁
            $this->redis->watch($key);
            $this->redis->multi();
            $this->redis->set($key, $newValue);
            $result = $this->redis->exec();
            $retryCount++;
        } while (!$result && $retryCount < 10);
        return $result;
    }
}

使用分布式系统（etcd/Consul）

<?php
// 使用 etcd 实现分布式锁和线性一致性
class DistributedConsensus {
    private $client;
    public function __construct($client) {
        $this->client = $client;
    }
    // 选举主节点（线性一致）
    public function electLeader($serviceName, $nodeId) {
        $key = "leader/{$serviceName}";
        $ttl = 30; // 租约时长
        // 尝试获取分布式锁
        try {
            $this->client->grant($ttl);
            $this->client->put($key, $nodeId, ['lease' => $ttl]);
            return true;
        } catch (Exception $e) {
            return false;
        }
    }
    // 读取最新配置（线性一致）
    public function getLatestConfig($configKey) {
        // 使用 etcd 的线性一致读
        return $this->client->get($configKey, ['quorum' => true]);
    }
}

性能优化建议

<?php
class OptimizedLinearizability {
    // 批处理操作减少锁竞争
    public function batchIncrement(array $userIds) {
        $pipeline = $this->redis->pipeline();
        foreach ($userIds as $userId) {
            $pipeline->incr("counter:{$userId}");
        }
        return $pipeline->exec();
    }
    // 使用 Redis 计数器（本身是原子的）
    public function incrementCounter($userId) {
        // Redis 的 INCR 操作本身就是线性一致的
        return $this->redis->incr("counter:{$userId}");
    }
    // 缓存 + 数据库一致性
    public function writeThrough($key, $value) {
        // 先写数据库
        $this->pdo->prepare(
            "INSERT INTO cache_store (k, v) VALUES (?, ?) ON CONFLICT (k) DO UPDATE SET v = EXCLUDED.v"
        )->execute([$key, $value]);
        // 再更新缓存
        $this->redis->set($key, $value);
        // 确保缓存与数据库一致
        return true;
    }
}