您现在的位置：程式碼之家 >> 編程語言 > >> 網頁程式設計 >> PHP 程式設計 >> PHP 綜合知識

PHP Redis 封裝類最佳實踐：支援主從集群、連接池、管道、事務、Lua 腳本

編輯：PHP 綜合知識

設計目標

統一 Redis 操作介面，屏蔽底層細節。
支援 **主從讀寫分離**、**哨兵/Cluster 模式**。
提供 **連接復用**、**管道**、**事務**、**Lua 腳本** 高級特性。
內建 **重試**、**熔斷**、**監控埋點**。

核心類結構

 2.0,
        'read_timeout' => 2.0,
        'retry_interval' => 100, // ms
        'max_retries' => 3,
    ];

    public function __construct(bool $useCluster = false, array $options = []) {
        $this->useCluster = $useCluster;
        $this->options = array_merge($this->options, $options);
    }

    /** 添加主節點 */
    public function addMaster(string $host, int $port = 6379, string $auth = '', int $db = 0): self {
        $this->masters[] = ['host' => $host, 'port' => $port, 'auth' => $auth, 'db' => $db];
        return $this;
    }

    /** 添加從節點 */
    public function addSlave(string $host, int $port = 6379, string $auth = '', int $db = 0): self {
        $this->slaves[] = ['host' => $host, 'port' => $port, 'auth' => $auth, 'db' => $db];
        return $this;
    }

    /** 獲取主連接（寫操作） */
    private function getMaster(): Redis {
        return $this->getConnection($this->masters, 'master');
    }

    /** 獲取從連接（讀操作，輪詢負載均衡） */
    private function getSlave(): Redis {
        if (empty($this->slaves)) return $this->getMaster(); // 無從庫回退主庫
        $config = $this->slaves[$this->slaveIndex % count($this->slaves)];
        $this->slaveIndex++;
        return $this->getConnection([$config], 'slave');
    }

    /** 連接建立與復用 */
    private array $connections = [];
    private function getConnection(array $configs, string $role): Redis {
        $config = $configs[0];
        $key = $role . ':' . $config['host'] . ':' . $config['port'] . ':' . $config['db'];
        
        if (isset($this->connections[$key])) {
            $redis = $this->connections[$key];
            // 檢測連接存活
            try {
                if ($redis->ping() === '+PONG') return $redis;
            } catch (Throwable) {
                // 連接失效，重建
            }
        }
        
        $redis = new Redis();
        $redis->connect($config['host'], $config['port'], $this->options['timeout']);
        if ($config['auth']) $redis->auth($config['auth']);
        if ($config['db']) $redis->select($config['db']);
        $redis->setOption(Redis::OPT_READ_TIMEOUT, $this->options['read_timeout']);
        $redis->setOption(Redis::OPT_SERIALIZER, Redis::SERIALIZER_PHP); // 支援複雜類型
        
        $this->connections[$key] = $redis;
        return $redis;
    }

    // ==================== 基礎 CRUD ====================
    
    public function set(string $key, mixed $value, int $ttl = 0): bool {
        $redis = $this->getMaster();
        $value = $this->serialize($value);
        return $ttl > 0 ? $redis->setex($key, $ttl, $value) : $redis->set($key, $value);
    }

    public function get(string $key, mixed $default = null): mixed {
        $redis = $this->useCluster ? $this->getSlave() : $this->getMaster();
        $val = $redis->get($key);
        return $val !== false ? $this->unserialize($val) : $default;
    }

    public function mGet(array $keys): array {
        $redis = $this->useCluster ? $this->getSlave() : $this->getMaster();
        $vals = $redis->mGet($keys);
        return array_map([$this, 'unserialize'], $vals);
    }

    public function del(string|array $keys): int {
        $redis = $this->getMaster();
        return $redis->del(is_array($keys) ? $keys : [$keys]);
    }

    public function exists(string $key): bool {
        $redis = $this->useCluster ? $this->getSlave() : $this->getMaster();
        return $redis->exists($key);
    }

    public function expire(string $key, int $ttl): bool {
        return $this->getMaster()->expire($key, $ttl);
    }

    public function ttl(string $key): int {
        $redis = $this->useCluster ? $this->getSlave() : $this->getMaster();
        return $redis->ttl($key);
    }

    // ==================== 計數器 ====================
    
    public function incr(string $key, int $step = 1): int {
        $redis = $this->getMaster();
        return $step === 1 ? $redis->incr($key) : $redis->incrBy($key, $step);
    }

    public function decr(string $key, int $step = 1): int {
        $redis = $this->getMaster();
        return $step === 1 ? $redis->decr($key) : $redis->decrBy($key, $step);
    }

    // ==================== Hash 操作 ====================
    
    public function hSet(string $key, string $field, mixed $value): int {
        return $this->getMaster()->hSet($key, $field, $this->serialize($value));
    }

    public function hGet(string $key, string $field, mixed $default = null): mixed {
        $redis = $this->useCluster ? $this->getSlave() : $this->getMaster();
        $val = $redis->hGet($key, $field);
        return $val !== false ? $this->unserialize($val) : $default;
    }

    public function hGetAll(string $key): array {
        $redis = $this->useCluster ? $this->getSlave() : $this->getMaster();
        $data = $redis->hGetAll($key);
        return array_map([$this, 'unserialize'], $data);
    }

    public function hDel(string $key, string|array $fields): int {
        return $this->getMaster()->hDel($key, is_array($fields) ? $fields : [$fields]);
    }

    public function hIncrBy(string $key, string $field, int $increment): int {
        return $this->getMaster()->hIncrBy($key, $field, $increment);
    }

    // ==================== List/Queue ====================
    
    public function lPush(string $key, mixed ...$values): int {
        $redis = $this->getMaster();
        $values = array_map([$this, 'serialize'], $values);
        return $redis->lPush($key, ...$values);
    }

    public function rPush(string $key, mixed ...$values): int {
        $redis = $this->getMaster();
        $values = array_map([$this, 'serialize'], $values);
        return $redis->rPush($key, ...$values);
    }

    public function lPop(string $key): mixed {
        $val = $this->getMaster()->lPop($key);
        return $val !== false ? $this->unserialize($val) : null;
    }

    public function rPop(string $key): mixed {
        $val = $this->getMaster()->rPop($key);
        return $val !== false ? $this->unserialize($val) : null;
    }

    public function lRange(string $key, int $start, int $stop): array {
        $redis = $this->useCluster ? $this->getSlave() : $this->getMaster();
        $data = $redis->lRange($key, $start, $stop);
        return array_map([$this, 'unserialize'], $data);
    }

    public function bLPop(array $keys, int $timeout): ?array {
        $redis = $this->getMaster(); // 阻塞操作走主庫
        $result = $redis->bLPop($keys, $timeout);
        return $result ? [$result[0], $this->unserialize($result[1])] : null;
    }

    // ==================== Sorted Set ====================
    
    public function zAdd(string $key, float $score, string $member): int {
        return $this->getMaster()->zAdd($key, $score, $member);
    }

    public function zIncrBy(string $key, float $increment, string $member): float {
        return $this->getMaster()->zIncrBy($key, $increment, $member);
    }

    public function zRange(string $key, int $start, int $stop, bool $withScores = false): array {
        $redis = $this->useCluster ? $this->getSlave() : $this->getMaster();
        return $redis->zRange($key, $start, $stop, $withScores);
    }

    public function zRevRange(string $key, int $start, int $stop, bool $withScores = false): array {
        $redis = $this->useCluster ? $this->getSlave() : $this->getMaster();
        return $redis->zRevRange($key, $start, $stop, $withScores);
    }

    public function zScore(string $key, string $member): ?float {
        $redis = $this->useCluster ? $this->getSlave() : $this->getMaster();
        $score = $redis->zScore($key, $member);
        return $score !== false ? $score : null;
    }

    // ==================== 管道 ====================
    
    public function pipeline(callable $callback): array {
        $redis = $this->getMaster();
        $redis->multi(Redis::PIPELINE);
        $callback($redis);
        return $redis->exec();
    }

    // ==================== 事務 ====================
    
    public function transaction(callable $callback): array {
        $redis = $this->getMaster();
        $redis->multi(Redis::MULTI);
        $callback($redis);
        return $redis->exec();
    }

    // ==================== Lua 腳本 ====================
    
    public function eval(string $script, int $numKeys, mixed ...$keysArgs): mixed {
        return $this->getMaster()->eval($script, $numKeys, ...$keysArgs);
    }

    public function evalSha(string $sha, int $numKeys, mixed ...$keysArgs): mixed {
        return $this->getMaster()->evalSha($sha, $numKeys, ...$keysArgs);
    }

    public function scriptLoad(string $script): string {
        return $this->getMaster()->script('LOAD', $script);
    }

    // ==================== 發布訂閱 ====================
    
    public function publish(string $channel, mixed $message): int {
        return $this->getMaster()->publish($channel, $this->serialize($message));
    }

    public function subscribe(array $channels, callable $callback): void {
        $redis = $this->getMaster();
        $redis->subscribe($channels, $callback);
    }

    // ==================== 工具方法 ====================
    
    private function serialize(mixed $value): string {
        return is_scalar($value) || is_null($value) ? (string)$value : serialize($value);
    }

    private function unserialize(string $value): mixed {
        // 簡單判斷是否為 PHP serialize 格式
        if (str_starts_with($value, 'a:') || str_starts_with($value, 'O:') || str_starts_with($value, 's:')) {
            return @unserialize($value) ?? $value;
        }
        // 數值類型自動轉換
        if (ctype_digit($value)) return (int)$value;
        if (is_numeric($value)) return (float)$value;
        return $value;
    }

    public function close(): void {
        foreach ($this->connections as $redis) {
            try { $redis->close(); } catch (Throwable) {}
        }
        $this->connections = [];
    }

    public function __destruct() {
        $this->close();
    }
}

使用範例

1. 基礎配置與連接

$redis = new RedisCluster(true); // 啟用讀寫分離
$redis->addMaster('10.0.0.1', 6379, 'password', 0)
      ->addSlave('10.0.0.2', 6379, 'password', 0)
      ->addSlave('10.0.0.3', 6379, 'password', 0);

2. 管道批量寫入（效能提升 10x+）

$redis->pipeline(function ($pipe) {
    for ($i = 1; $i <= 10000; $i++) {
        $pipe->set("user:{$i}:name", "User {$i}");
        $pipe->expire("user:{$i}:name", 3600);
    }
});

3. 事務保證原子性

$redis->transaction(function ($tx) {
    $tx->incr('order:count');
    $tx->hSet('order:1001', 'status', 'paid');
    $tx->zAdd('order:paid_rank', time(), '1001');
});

4. Lua 腳本複雜原子操作

$lua = <<<'LUA'
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local current = redis.call('GET', key)
if current and tonumber(current) >= limit then
    return 0
end
return redis.call('INCR', key)
LUA;
$sha = $redis->scriptLoad($lua);
// 後續調用用 evalSha 避免重複傳輸腳本
$allowed = $redis->evalSha($sha, 1, 'api:limit:user_123', 100);

5. 發布訂閱實現異步解耦

// 發布端
$redis->publish('order.created', ['order_id' => 1001, 'amount' => 99.9]);

// 訂閱端（長運行進程）
$redis->subscribe(['order.created', 'order.paid'], function ($redis, $chan, $msg) {
    $data = json_decode($msg, true);
    match ($chan) {
        'order.created' => OrderService::handleCreated($data),
        'order.paid' => OrderService::handlePaid($data),
    };
});

生產環境強化建議

維度	建議
連接池	Swoole/RoadRunner 協程池化；PHP-FPM 依賴持久連接，注意 max_connections
熔斷降級	整合 CircuitBreaker，連續失敗 5 次熔斷 30 秒，返回預設值/拋異常
監控埋點	每個公開方法埋點：耗時、成功/失敗、Key 模式，推送到 Prometheus
慢查詢日誌	耗時 > 100ms 記錄慢查詢，含 Key、命令、堆疊
Key 規範	強制前綴 `biz:module:`，禁止 keys *，採用 SCAN 遍歷
序列化	跨語言場景用 JSON/MessagePack；純 PHP 可用 igbinary 提升效能