怎样在PHP项目中实现分库分表？

wen java案例 2026-07-23 159

本文目录导读：

怎样在PHP项目中实现分库分表？

分库分表策略
完整的分库分表实现
使用ORM框架实现（以ThinkPHP为例）
使用中间件方案（如ShardingSphere）
操作注意事项
性能优化建议

在PHP项目中实现分库分表是一个比较复杂的架构设计,主要解决单库单表数据量过大导致的性能瓶颈，下面是完整的实现方案：

分库分表策略

垂直分库

// 按业务模块拆分到不同数据库
class DBConfig {
    const DB_USER = [
        'host' => '192.168.1.1',
        'dbname' => 'user_db'
    ];
    const DB_ORDER = [
        'host' => '192.168.1.2', 
        'dbname' => 'order_db'
    ];
    const DB_PRODUCT = [
        'host' => '192.168.1.3',
        'dbname' => 'product_db'
    ];
}

水平分表

// 常用分片算法
class ShardingStrategy {
    /**
     * 取模分片（最常用）
     */
    public static function modSharding($userId, $tableCount = 16) {
        return $userId % $tableCount;
    }
    /**
     * 哈希分片
     */
    public static function hashSharding($key, $tableCount = 16) {
        return crc32($key) % $tableCount;
    }
    /**
     * 范围分片
     */
    public static function rangeSharding($userId) {
        if ($userId <= 10000) return 'user_0';
        if ($userId <= 20000) return 'user_1';
        return 'user_2';
    }
    /**
     * 日期分片（适用于日志等）
     */
    public static function dateSharding($date) {
        $month = date('Ym', strtotime($date));
        return "log_{$month}";
    }
}

完整的分库分表实现

分库分表中间件类

<?php
class DatabaseSharding {
    private $connections = [];
    private $shardingConfig;
    public function __construct($config) {
        $this->shardingConfig = $config;
    }
    /**
     * 获取连接的数据库
     */
    public function getConnection($shardKey) {
        $dbIndex = $this->getDbIndex($shardKey);
        if (!isset($this->connections[$dbIndex])) {
            $config = $this->shardingConfig['databases'][$dbIndex];
            $this->connections[$dbIndex] = $this->createConnection($config);
        }
        return $this->connections[$dbIndex];
    }
    /**
     * 获取表名
     */
    public function getTableName($baseTable, $shardKey) {
        $tableIndex = $this->getTableIndex($shardKey);
        return "{$baseTable}_{$tableIndex}";
    }
    /**
     * 计算数据库索引
     */
    private function getDbIndex($shardKey) {
        return $shardKey % $this->shardingConfig['db_count'];
    }
    /**
     * 计算表索引
     */
    private function getTableIndex($shardKey) {
        return $shardKey % $this->shardingConfig['table_per_db'];
    }
    /**
     * 创建数据库连接
     */
    private function createConnection($config) {
        $dsn = "mysql:host={$config['host']};port={$config['port']};dbname={$config['dbname']};charset=utf8mb4";
        try {
            $pdo = new PDO($dsn, $config['username'], $config['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());
        }
    }
    /**
     * 执行查询
     */
    public function query($sql, $params = [], $shardKey = null) {
        if ($shardKey === null) {
            // 全库扫描（不推荐，仅用于特殊场景）
            return $this->queryAllShards($sql, $params);
        }
        $connection = $this->getConnection($shardKey);
        $tableName = $this->getTableName($this->extractTableName($sql), $shardKey);
        $sql = str_replace(':table:', $tableName, $sql);
        $stmt = $connection->prepare($sql);
        $stmt->execute($params);
        return $stmt->fetchAll();
    }
    /**
     * 全库扫描（跨分片查询）
     */
    private function queryAllShards($sql, $params) {
        $results = [];
        for ($i = 0; $i < $this->shardingConfig['db_count']; $i++) {
            $connection = $this->getConnectionByIndex($i);
            for ($j = 0; $j < $this->shardingConfig['table_per_db']; $j++) {
                $tableName = "{$this->extractTableName($sql)}_{$j}";
                $currentSql = str_replace(':table:', $tableName, $sql);
                $stmt = $connection->prepare($currentSql);
                $stmt->execute($params);
                $results = array_merge($results, $stmt->fetchAll());
            }
        }
        return $results;
    }
}

用户表分库分表示例

<?php
class UserService {
    private $sharding;
    public function __construct() {
        $config = [
            'db_count' => 4, // 4个数据库实例
            'table_per_db' => 4, // 每个库4张表
            'databases' => [
                [
                    'host' => '192.168.1.1',
                    'port' => 3306,
                    'dbname' => 'user_db_0',
                    'username' => 'root',
                    'password' => 'password'
                ],
                [
                    'host' => '192.168.1.1',
                    'port' => 3306,
                    'dbname' => 'user_db_1',
                    'username' => 'root',
                    'password' => 'password'
                ],
                // ... 更多数据库配置
            ]
        ];
        $this->sharding = new DatabaseSharding($config);
    }
    /**
     * 获取用户信息
     */
    public function getUser($userId) {
        $sql = "SELECT * FROM :table: WHERE user_id = ?";
        $result = $this->sharding->query($sql, [$userId], $userId);
        return $result[0] ?? null;
    }
    /**
     * 添加用户
     */
    public function addUser($userData) {
        $userId = $userData['user_id'];
        $sql = "INSERT INTO :table: (user_id, username, email) VALUES (?, ?, ?)";
        return $this->sharding->query($sql, [
            $userId,
            $userData['username'],
            $userData['email']
        ], $userId);
    }
    /**
     * 批量查询用户（跨分片）
     */
    public function getUsersByCondition($condition, $value) {
        // 这种查询需要全表扫描，性能较差
        $sql = "SELECT * FROM :table: WHERE {$condition} = ?";
        return $this->sharding->query($sql, [$value]);
    }
}

使用ORM框架实现（以ThinkPHP为例）

<?php
// 配置文件 database.php
return [
    // 分库配置
    'connections' => [
        'user_db_0' => [
            'type' => 'mysql',
            'hostname' => '192.168.1.1',
            'database' => 'user_db_0',
        ],
        'user_db_1' => [
            'type' => 'mysql',
            'hostname' => '192.168.1.1', 
            'database' => 'user_db_1',
        ],
    ],
    // 分表规则
    'sharding_rules' => [
        'user' => [
            'type' => 'mod',
            'field' => 'user_id',
            'database_count' => 2,
            'table_count' => 4,
        ]
    ]
];
// 模型中使用
namespace app\model;
use think\Model;
class User extends Model
{
    // 定义分库分表规则
    protected function getShardingKey($userId) {
        return $userId % 2; // 分库键
    }
    protected function getTableName($userId) {
        $tableIndex = $userId % 4; // 分表键
        return "user_{$tableIndex}";
    }
    public function findByUserId($userId) {
        $dbKey = $this->getShardingKey($userId);
        $tableName = $this->getTableName($userId);
        // 切换到对应数据库
        $this->connection("user_db_{$dbKey}");
        // 设置表名
        $this->table($tableName);
        return $this->where('user_id', $userId)->find();
    }
}

使用中间件方案（如ShardingSphere）

配置ShardingSphere-Proxy

# config-sharding.yaml
schemaName: sharding_db
dataSources:
  ds_0:
    url: jdbc:mysql://192.168.1.1:3306/demo_ds_0
    username: root
    password: 
  ds_1:
    url: jdbc:mysql://192.168.1.2:3306/demo_ds_1
    username: root
    password: 
rules:
- !SHARDING
  tables:
    t_order:
      actualDataNodes: ds_${0..1}.t_order_${0..1}
      tableStrategy:
        standard:
          shardingColumn: order_id
          shardingAlgorithmName: t_order_inline
      keyGenerateStrategy:
        column: order_id
        keyGeneratorName: snowflake
  shardingAlgorithms:
    t_order_inline:
      type: INLINE
      props:
        algorithm-expression: ds_${order_id % 2}.t_order_${order_id % 2}

操作注意事项

分片键选择原则

// 好的分片键：用户ID、订单ID
// 好的分片键：高频查询字段
// 避免的分片键：状态字段、时间字段（除非特定场景）
// 避免的分片键：非唯一字段

跨分片查询处理

// 1. 避免跨分片JOIN
// 2. 使用全局表（配置表不拆分）
// 3. 使用Elasticsearch等搜索引擎
// 4. 数据汇总到中间表
// 示例：用户+订单关联查询
class CrossShardQuery {
    // 方法1：应用层聚合
    public function getUserOrders($userId) {
        $userInfo = $this->userService->getUser($userId);
        $orderInfo = $this->orderService->getOrdersByUserId($userId);
        return array_merge($userInfo, ['orders' => $orderInfo]);
    }
    // 方法2：冗余字段
    public function getOrderWithUserName($orderId) {
        $order = $this->orderService->getOrder($orderId);
        $user = $this->userService->getUser($order['user_id']);
        $order['username'] = $user['username'];
        return $order;
    }
}

数据迁移方案

-- 1. 停机迁移
-- 2. 双写方案
-- 3. 定时任务增量同步
-- 双写示例
BEGIN;
-- 写新库
INSERT INTO new_user_db.user_1 (user_id, username) VALUES (?, ?);
-- 写旧库
INSERT INTO old_user_db.user (user_id, username) VALUES (?, ?);
COMMIT;