本文目录导读：
在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;
性能优化建议
- 连接池：使用连接池管理数据库连接
- 缓存层：Redis缓存热点数据
- 读写分离：主库写，从库读
- 监控告警：监控各分片数据量、查询耗时
这样实现后,可以很好地解决大数据量下的性能问题，同时保持了代码的相对简单和可维护性。
