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

如何用PHP项目实现主从同步？

wen java案例 2

本文目录导读：

如何用PHP项目实现主从同步？

  1. 数据库主从配置
  2. PHP应用层读写分离实现
  3. 高级优化方案
  4. 注意事项

在PHP项目中实现MySQL主从同步，主要涉及数据库层面的配置和应用层面的读写分离,以下是完整的实现步骤：

数据库主从配置

主服务器配置 (Master)

# /etc/mysql/mysql.conf.d/mysqld.cnf
[mysqld]
server-id = 1
log_bin = /var/log/mysql/mysql-bin.log
binlog_do_db = your_database_name  # 要同步的数据库
binlog_ignore_db = mysql           # 忽略的数据库

创建复制用户：

CREATE USER 'replication'@'%' IDENTIFIED BY 'password';
GRANT REPLICATION SLAVE ON *.* TO 'replication'@'%';
FLUSH PRIVILEGES;

查看主服务器状态：

SHOW MASTER STATUS;
-- 记录 File 和 Position 值

从服务器配置 (Slave)

# /etc/mysql/mysql.conf.d/mysqld.cnf
[mysqld]
server-id = 2
relay_log = /var/log/mysql/mysql-relay-bin.log
log_bin = /var/log/mysql/mysql-bin.log

配置从服务器连接主服务器：

CHANGE MASTER TO
MASTER_HOST='master_ip',
MASTER_USER='replication',
MASTER_PASSWORD='password',
MASTER_LOG_FILE='mysql-bin.000001',  -- 从主服务器获取
MASTER_LOG_POS=123;                  -- 从主服务器获取
START SLAVE;
-- 检查状态
SHOW SLAVE STATUS\G
-- 确保 Slave_IO_Running 和 Slave_SQL_Running 都为 Yes

PHP应用层读写分离实现

基础连接配置

<?php
// config/database.php
return [
    'master' => [
        'host' => '192.168.1.100',
        'port' => 3306,
        'database' => 'your_database',
        'username' => 'write_user',
        'password' => 'write_password',
        'charset' => 'utf8mb4',
    ],
    'slaves' => [
        [
            'host' => '192.168.1.101',
            'port' => 3306,
            'database' => 'your_database',
            'username' => 'read_user',
            'password' => 'read_password',
            'charset' => 'utf8mb4',
        ],
        [
            'host' => '192.168.1.102',
            'port' => 3306,
            'database' => 'your_database',
            'username' => 'read_user',
            'password' => 'read_password',
            'charset' => 'utf8mb4',
        ],
    ],
];

PDO连接管理器

<?php
// lib/DatabaseManager.php
class DatabaseManager
{
    private static $instances = [];
    private $config;
    private $masterConnection;
    private $slaveConnection;
    public function __construct($config)
    {
        $this->config = $config;
    }
    // 获取主库连接（写操作）
    public function getMasterConnection()
    {
        if (!$this->masterConnection) {
            $this->masterConnection = $this->createConnection($this->config['master']);
        }
        return $this->masterConnection;
    }
    // 获取从库连接（读操作）
    public function getSlaveConnection()
    {
        // 随机选择一个从库
        $slaves = $this->config['slaves'];
        $slave = $slaves[array_rand($slaves)];
        $key = $slave['host'] . ':' . $slave['port'];
        if (!isset(self::$instances[$key])) {
            self::$instances[$key] = $this->createConnection($slave);
        }
        return self::$instances[$key];
    }
    private function createConnection($dbConfig)
    {
        $dsn = sprintf(
            "mysql:host=%s;port=%d;dbname=%s;charset=%s",
            $dbConfig['host'],
            $dbConfig['port'],
            $dbConfig['database'],
            $dbConfig['charset']
        );
        try {
            $pdo = new PDO($dsn, $dbConfig['username'], $dbConfig['password'], [
                PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
                PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
                PDO::ATTR_EMULATE_PREPARES => false,
            ]);
            return $pdo;
        } catch (PDOException $e) {
            throw new Exception("Database connection failed: " . $e->getMessage());
        }
    }
}

读写分离的查询构建器

<?php
// lib/QueryBuilder.php
class QueryBuilder
{
    private $dbManager;
    public function __construct(DatabaseManager $dbManager)
    {
        $this->dbManager = $dbManager;
    }
    // 写操作 - 使用主库
    public function insert($table, $data)
    {
        $pdo = $this->dbManager->getMasterConnection();
        $columns = implode(', ', array_keys($data));
        $placeholders = ':' . implode(', :', array_keys($data));
        $sql = "INSERT INTO {$table} ({$columns}) VALUES ({$placeholders})";
        $stmt = $pdo->prepare($sql);
        return $stmt->execute($data);
    }
    // 读操作 - 使用从库
    public function select($table, $conditions = [], $fields = '*')
    {
        $pdo = $this->dbManager->getSlaveConnection();
        $sql = "SELECT {$fields} FROM {$table}";
        if (!empty($conditions)) {
            $where = [];
            foreach ($conditions as $key => $value) {
                $where[] = "{$key} = :{$key}";
            }
            $sql .= " WHERE " . implode(' AND ', $where);
        }
        $stmt = $pdo->prepare($sql);
        $stmt->execute($conditions);
        return $stmt->fetchAll();
    }
    // 更新操作 - 使用主库
    public function update($table, $data, $conditions)
    {
        $pdo = $this->dbManager->getMasterConnection();
        $setParts = [];
        foreach ($data as $key => $value) {
            $setParts[] = "{$key} = :set_{$key}";
        }
        $whereParts = [];
        foreach ($conditions as $key => $value) {
            $whereParts[] = "{$key} = :where_{$key}";
        }
        $sql = "UPDATE {$table} SET " . implode(', ', $setParts) 
             . " WHERE " . implode(' AND ', $whereParts);
        $params = [];
        foreach ($data as $key => $value) {
            $params["set_{$key}"] = $value;
        }
        foreach ($conditions as $key => $value) {
            $params["where_{$key}"] = $value;
        }
        $stmt = $pdo->prepare($sql);
        return $stmt->execute($params);
    }
    // 删除操作 - 使用主库
    public function delete($table, $conditions)
    {
        $pdo = $this->dbManager->getMasterConnection();
        $whereParts = [];
        foreach ($conditions as $key => $value) {
            $whereParts[] = "{$key} = :{$key}";
        }
        $sql = "DELETE FROM {$table} WHERE " . implode(' AND ', $whereParts);
        $stmt = $pdo->prepare($sql);
        return $stmt->execute($conditions);
    }
}

使用方法

<?php
// index.php
require_once 'config/database.php';
require_once 'lib/DatabaseManager.php';
require_once 'lib/QueryBuilder.php';
// 初始化
$config = include 'config/database.php';
$dbManager = new DatabaseManager($config);
$queryBuilder = new QueryBuilder($dbManager);
// 写操作 - 自动使用主库
$userId = $queryBuilder->insert('users', [
    'name' => '张三',
    'email' => 'zhangsan@example.com',
    'created_at' => date('Y-m-d H:i:s')
]);
// 读操作 - 自动使用从库
$users = $queryBuilder->select('users', ['status' => 1], 'id, name, email');
// 更新操作 - 自动使用主库
$queryBuilder->update('users', 
    ['status' => 2, 'updated_at' => date('Y-m-d H:i:s')],
    ['id' => $userId]
);
// 删除操作 - 自动使用主库
$queryBuilder->delete('users', ['id' => $userId]);

高级优化方案

连接池实现（使用Swoole）

<?php
// lib/ConnectionPool.php
class ConnectionPool
{
    private $pool;
    private $maxConnections;
    private $connectionConfig;
    public function __construct($config, $maxConnections = 10)
    {
        $this->connectionConfig = $config;
        $this->maxConnections = $maxConnections;
        $this->pool = new SplQueue();
    }
    public function getConnection()
    {
        if ($this->pool->count() > 0) {
            return $this->pool->dequeue();
        }
        return $this->createConnection();
    }
    public function releaseConnection($connection)
    {
        if ($this->pool->count() < $this->maxConnections) {
            $this->pool->enqueue($connection);
        } else {
            $connection = null;
        }
    }
    private function createConnection()
    {
        // 创建连接逻辑
    }
}

主从延迟处理

<?php
// 对于关键数据的读取，强制使用主库
class CriticalDataReader
{
    private $dbManager;
    public function getLatestUserData($userId)
    {
        // 某些场景下，读最新数据需要强制用主库
        $pdo = $this->dbManager->getMasterConnection();
        $stmt = $pdo->prepare("SELECT * FROM users WHERE id = ?");
        $stmt->execute([$userId]);
        return $stmt->fetch();
    }
}

监控和告警

<?php
// lib/Monitor.php
class Monitor
{
    public function checkReplicationStatus()
    {
        // 检查主从复制状态
        $pdo = $this->dbManager->getSlaveConnection();
        $stmt = $pdo->query("SHOW SLAVE STATUS");
        $status = $stmt->fetch();
        if ($status['Slave_IO_Running'] !== 'Yes' || 
            $status['Slave_SQL_Running'] !== 'Yes') {
            // 发送告警
            $this->sendAlert('Replication is down!');
        }
        // 检查延迟
        $secondsBehind = $status['Seconds_Behind_Master'];
        if ($secondsBehind > 60) {
            $this->sendAlert("Replication lag: {$secondsBehind} seconds");
        }
    }
    private function sendAlert($message)
    {
        // 发送邮件、短信或使用告警系统
        error_log($message);
        // 实际项目中可以使用第三方服务
    }
}

注意事项

  1. 事务处理：事务中的读写都应该使用主库
  2. 主从延迟：对实时性要求高的场景，考虑强制使用主库
  3. 连接管理：使用连接池减少连接开销
  4. 负载均衡：合理分配读请求到多个从库
  5. 监控告警：定期检查主从同步状态
  6. 容错处理：从库故障时自动切换到其他从库或主库

这个实现方案提供了完整的MySQL主从同步解决方案，从数据库配置到PHP应用层实现，涵盖了读写分离、连接管理、延迟处理等核心功能。

上一篇PHP项目怎么实现数据库集群？

下一篇PHP项目如何实现读写分离？

相关文章

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