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

怎样在PHP项目中实现断线重连？

wen java案例 2

本文目录导读：

怎样在PHP项目中实现断线重连？

  1. 数据库连接断线重连
  2. 常驻进程场景（Swoole/Workerman）
  3. WebSocket长连接断线重连
  4. 最佳实践建议
  5. 注意事项

在PHP项目中实现断线重连主要涉及数据库连接和长连接（如WebSocket） 两种情况，以下是基于PHP-FPM短连接模型和常驻进程场景的完整实现方案。

数据库连接断线重连

PHP-FPM通常是短连接，但连接池或常驻进程场景（如Swoole）需要重连机制。

PDO重连方案

<?php
class DatabaseReconnect
{
    private $pdo;
    private $config;
    private $maxRetries = 3;
    private $retryDelay = 100; // 毫秒
    public function __construct(array $config)
    {
        $this->config = $config;
        $this->connect();
    }
    private function connect(): void
    {
        try {
            $dsn = sprintf(
                'mysql:host=%s;port=%d;dbname=%s;charset=utf8mb4',
                $this->config['host'],
                $this->config['port'] ?? 3306,
                $this->config['database']
            );
            $this->pdo = new PDO($dsn, $this->config['username'], $this->config['password'], [
                PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
                PDO::ATTR_EMULATE_PREPARES => false,
                PDO::ATTR_TIMEOUT => 5, // 超时5秒
            ]);
        } catch (PDOException $e) {
            throw new RuntimeException('Database connection failed: ' . $e->getMessage());
        }
    }
    // 核心：重试执行SQL
    public function query(string $sql, array $params = [])
    {
        $attempts = 0;
        while ($attempts < $this->maxRetries) {
            try {
                $stmt = $this->pdo->prepare($sql);
                $stmt->execute($params);
                return $stmt;
            } catch (PDOException $e) {
                // 只对连接相关错误重试
                if ($this->isConnectionError($e)) {
                    $attempts++;
                    if ($attempts >= $this->maxRetries) {
                        throw $e; // 重试耗尽，抛出原始异常
                    }
                    // 等待后重连
                    usleep($this->retryDelay * 1000);
                    $this->reconnect();
                } else {
                    throw $e; // 其他SQL错误直接抛出
                }
            }
        }
    }
    private function isConnectionError(PDOException $e): bool
    {
        $errorMsg = $e->getMessage();
        $connectionErrors = [
            'server has gone away',
            'lost connection',
            'Error while reading',
            'MySQL server has gone away',
            'Connection refused',
            'broken pipe',
        ];
        foreach ($connectionErrors as $keyword) {
            if (stripos($errorMsg, $keyword) !== false) {
                return true;
            }
        }
        return false;
    }
    private function reconnect(): void
    {
        $this->pdo = null; // 关闭旧连接
        $this->connect();
    }
}
// 使用示例
$config = [
    'host'     => 'localhost',
    'port'     => 3306,
    'database' => 'test',
    'username' => 'root',
    'password' => 'secret',
];
$db = new DatabaseReconnect($config);
$result = $db->query('SELECT * FROM users WHERE id = ?', [1]);

Mysqli重连方案

<?php
class MysqliReconnect
{
    private $mysqli;
    private $config;
    private $maxRetries = 3;
    public function __construct(array $config)
    {
        $this->config = $config;
        $this->connect();
    }
    private function connect(): void
    {
        $this->mysqli = new mysqli(
            $this->config['host'],
            $this->config['username'],
            $this->config['password'],
            $this->config['database'],
            $this->config['port'] ?? 3306
        );
        if ($this->mysqli->connect_error) {
            throw new RuntimeException('Connection failed: ' . $this->mysqli->connect_error);
        }
        $this->mysqli->set_charset('utf8mb4');
    }
    public function query(string $sql)
    {
        $attempts = 0;
        while ($attempts < $this->maxRetries) {
            $result = $this->mysqli->query($sql);
            if ($result === false) {
                // 检测是否连接断开
                if ($this->mysqli->errno === 2006 || $this->mysqli->errno === 2013) {
                    $attempts++;
                    $this->reconnect();
                    continue;
                }
                throw new RuntimeException('Query error: ' . $this->mysqli->error);
            }
            return $result;
        }
    }
    private function reconnect(): void
    {
        $this->mysqli->close();
        $this->connect();
    }
}

常驻进程场景（Swoole/Workerman）

Swoole协程MySQL断线重连

<?php
use Swoole\Coroutine\MySQL;
use Swoole\Coroutine;
class SwooleMySQLReconnect
{
    private $mysql;
    private $config;
    private $lastPingTime = 0;
    private $pingInterval = 60; // 秒
    public function __construct(array $config)
    {
        $this->config = $config;
        $this->connect();
    }
    private function connect(): void
    {
        $this->mysql = new MySQL();
        $connected = $this->mysql->connect([
            'host'     => $this->config['host'],
            'port'     => $this->config['port'] ?? 3306,
            'user'     => $this->config['username'],
            'password' => $this->config['password'],
            'database' => $this->config['database'],
            'charset'  => 'utf8mb4',
            'timeout'  => 5, // 连接超时
        ]);
        if (!$connected) {
            throw new RuntimeException('Connect failed: ' . $this->mysql->connect_error);
        }
    }
    public function query(string $sql, array $params = [])
    {
        $this->ping(); // 自动检测连接
        $stmt = $this->mysql->prepare($sql);
        if (!$stmt) {
            // 如果prepare失败，可能是连接断开
            $this->reconnect();
            $stmt = $this->mysql->prepare($sql);
        }
        if (!$stmt) {
            throw new RuntimeException('Prepare failed: ' . $this->mysql->error);
        }
        $result = $stmt->execute($params);
        if ($result === false) {
            // 如果是连接错误，重连后重试一次
            if ($this->mysql->errno === 2006 || $this->mysql->errno === 2013) {
                $this->reconnect();
                $stmt = $this->mysql->prepare($sql);
                $result = $stmt->execute($params);
            }
        }
        return $stmt->fetchAll();
    }
    private function ping(): void
    {
        $now = time();
        if ($now - $this->lastPingTime > $this->pingInterval) {
            if (!$this->mysql->query('SELECT 1')) {
                $this->reconnect();
            }
            $this->lastPingTime = $now;
        }
    }
    private function reconnect(): void
    {
        $this->mysql->close();
        $this->connect();
    }
}

Redis断线重连

<?php
class RedisReconnect
{
    private $redis;
    private $config;
    private $maxRetries = 3;
    public function __construct(array $config)
    {
        $this->config = $config;
        $this->connect();
    }
    private function connect(): void
    {
        $this->redis = new Redis();
        if (!$this->redis->connect($this->config['host'], $this->config['port'], 5)) {
            throw new RuntimeException('Redis connection failed');
        }
        if (!empty($this->config['password'])) {
            $this->redis->auth($this->config['password']);
        }
    }
    public function __call($method, $args)
    {
        $attempts = 0;
        while ($attempts < $this->maxRetries) {
            try {
                return call_user_func_array([$this->redis, $method], $args);
            } catch (RedisException $e) {
                // 检测是否连接断开
                if ($this->isConnectionError($e)) {
                    $attempts++;
                    $this->reconnect();
                    continue;
                }
                throw $e;
            }
        }
    }
    private function isConnectionError(RedisException $e): bool
    {
        $msg = $e->getMessage();
        return stripos($msg, 'connection lost') !== false 
            || stripos($msg, 'closed') !== false
            || stripos($msg, 'broken pipe') !== false;
    }
    private function reconnect(): void
    {
        $this->redis->close();
        $this->connect();
    }
}

WebSocket长连接断线重连

客户端JavaScript实现

class WebSocketReconnect {
    constructor(url, options = {}) {
        this.url = url;
        this.options = {
            maxRetries: 5,
            retryDelay: 1000,     // 初始延迟1秒
            retryDelayMax: 30000, // 最大延迟30秒
            ...options
        };
        this.retries = 0;
        this.ws = null;
        this.connect();
    }
    connect() {
        try {
            this.ws = new WebSocket(this.url);
            this.ws.onopen = () => {
                console.log('WebSocket connected');
                this.retries = 0; // 重置重试次数
                this.options.onOpen?.();
            };
            this.ws.onclose = (event) => {
                console.log('WebSocket disconnected:', event.code);
                this.options.onClose?.(event);
                // 非正常关闭时自动重连
                if (event.code !== 1000 && event.code !== 1001) {
                    this.reconnect();
                }
            };
            this.ws.onerror = (error) => {
                console.error('WebSocket error:', error);
                this.options.onError?.(error);
            };
            this.ws.onmessage = (message) => {
                this.options.onMessage?.(message);
            };
        } catch (error) {
            console.error('WebSocket connection failed:', error);
            this.reconnect();
        }
    }
    reconnect() {
        if (this.retries >= this.options.maxRetries) {
            console.error('Max retries reached, stop reconnecting');
            this.options.onMaxRetries?.();
            return;
        }
        this.retries++;
        // 指数退避算法
        const delay = Math.min(
            this.options.retryDelay * Math.pow(2, this.retries - 1),
            this.options.retryDelayMax
        );
        console.log(`Reconnecting in ${delay}ms (attempt ${this.retries}/${this.options.maxRetries})`);
        setTimeout(() => {
            this.connect();
        }, delay);
    }
    send(data) {
        if (this.ws && this.ws.readyState === WebSocket.OPEN) {
            this.ws.send(data);
        } else {
            console.warn('WebSocket not connected, message queued');
            this.options.onMessageQueued?.(data);
        }
    }
    close() {
        this.retries = this.options.maxRetries; // 防止重连
        this.ws?.close(1000, 'Client closed');
    }
}

服务端PHP WebSocket重连处理

<?php
// 使用Workerman或Swoole实现
// 这里以Workerman为例
use Workerman\Worker;
use Workerman\Connection\TcpConnection;
require_once __DIR__ . '/vendor/autoload.php';
$ws_worker = new Worker('websocket://0.0.0.0:8080');
$ws_worker->count = 4;
// 连接管理
$connections = new SplObjectStorage();
$ws_worker->onConnect = function(TcpConnection $connection) use ($connections) {
    $connections->attach($connection);
    echo "New connection: {$connection->id}\n";
};
$ws_worker->onMessage = function(TcpConnection $connection, $data) {
    // 处理业务逻辑
    echo "Received: $data\n";
    // 模拟处理
    $response = json_encode(['status' => 'ok', 'data' => $data]);
    $connection->send($response);
};
$ws_worker->onClose = function(TcpConnection $connection) use ($connections) {
    $connections->detach($connection);
    echo "Connection closed: {$connection->id}\n";
    // 可以在这里保存断线状态，等待客户端重连
    // 保存用户会话到Redis，设置过期时间
};
// 运行worker
Worker::runAll();

最佳实践建议

连接池化

// 示例：使用Swoole连接池
$pool = new Swoole\Database\PDOPool(
    (new Swoole\Database\PDOConfig())
        ->withHost('localhost')
        ->withPort(3306)
        ->withDbName('test')
        ->withUsername('root')
        ->withPassword('secret')
        ->withTimeout(5),
    $size = 10 // 连接池大小
);
// 连接池会自动管理连接的创建和断开

心跳检测

  • 对长连接定期发送PING命令
  • 数据库：SELECT 1
  • Redis：PING
  • WebSocket：自定义心跳包

指数退避重试

function exponentialBackoff(int $attempt): int
{
    $baseDelay = 100; // 100ms
    $maxDelay = 30000; // 30s
    $delay = $baseDelay * pow(2, $attempt - 1);
    return min($delay, $maxDelay) + random_int(0, 50); // 加随机抖动
}

错误分类处理

  • 可重试错误：连接超时、服务器重启、网络闪断
  • 不可重试错误：SQL语法错误、权限不足、参数错误

监控告警

// 记录重连事件到日志
function logReconnect(string $type, int $attempt, string $error): void
{
    $log = sprintf(
        "[%s] %s reconnect attempt %d: %s",
        date('Y-m-d H:i:s'),
        $type,
        $attempt,
        $error
    );
    error_log($log, 3, '/var/log/php-reconnect.log');
    // 可以发送到监控系统
    // sendAlert($log);
}

注意事项

  1. 事务处理：重连后需要重新开始事务，已执行部分自动回滚
  2. 数据一致性：确保不会因为重连导致数据重复执行
  3. 资源释放：重连前一定要关闭旧连接，避免资源泄露
  4. 并发控制：常驻进程环境注意协程间的连接隔离
  5. 超时设置：合理设置连接超时、查询超时时间

通过以上方案,可以确保PHP应用在面对网络抖动、数据库重启等情况时保持稳定运行，根据实际场景选择合适的重连策略，并做好监控告警。

上一篇如何用PHP项目实现健康检查？

下一篇PHP项目如何实现故障转移？

相关文章

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