本文目录导读：
对于PHP项目中的跨分片查询优化聚合统计逻辑，这是一个典型的分布式数据库查询挑战,以下是我建议的优化策略和方法：
聚合逻辑分层处理
源端聚合（推荐）
在每个分片上先进行部分聚合,再汇总到协调节点：
// 分片查询示例
class ShardQueryOptimizer {
// 分片配置
private $shards = [
'shard1' => ['host' => '192.168.1.1', 'db' => 'db_shard_1'],
'shard2' => ['host' => '192.168.1.2', 'db' => 'db_shard_2']
];
public function optimizedAggregation($queryParams) {
$partialResults = [];
// 1. 在每个分片执行部分聚合
foreach ($this->shards as $shardName => $shardConfig) {
$partialResults[$shardName] = $this->queryShard(
$shardConfig,
"SELECT
DATE(created_at) as date,
COUNT(*) as count,
SUM(amount) as total_amount
FROM orders
WHERE status = 'completed'
GROUP BY DATE(created_at)"
);
}
// 2. 在应用层合并结果
return $this->mergePartialAggregations($partialResults);
}
private function mergePartialAggregations($results) {
$merged = [];
foreach ($results as $shardResults) {
foreach ($shardResults as $row) {
$key = $row['date'];
if (!isset($merged[$key])) {
$merged[$key] = [
'date' => $key,
'count' => 0,
'total_amount' => 0
];
}
$merged[$key]['count'] += $row['count'];
$merged[$key]['total_amount'] += $row['total_amount'];
}
}
return array_values($merged);
}
}
并行查询策略
使用多线程/协程并行查询
// 使用Swoole协程实现并行查询
class ParallelShardQuery {
public function parallelQuery($shardQueries) {
$results = [];
$wg = new \Swoole\Coroutine\WaitGroup();
foreach ($this->shards as $shardName => $shardConfig) {
$wg->add();
go(function() use ($shardName, $shardConfig, &$results, $wg) {
$results[$shardName] = $this->queryShard($shardConfig, $shardQueries[$shardName]);
$wg->done();
});
}
$wg->wait();
return $results;
}
}
预聚合与缓存策略
创建聚合表/物化视图
-- 在每个分片创建预聚合表
CREATE TABLE orders_daily_agg (
date DATE,
status VARCHAR(20),
order_count INT,
total_amount DECIMAL(15,2),
PRIMARY KEY (date, status)
);
-- 定期更新聚合数据
INSERT INTO orders_daily_agg (date, status, order_count, total_amount)
SELECT
DATE(created_at),
status,
COUNT(*),
SUM(amount)
FROM orders
WHERE created_at >= DATE_SUB(NOW(), INTERVAL 1 DAY)
GROUP BY DATE(created_at), status
ON DUPLICATE KEY UPDATE
order_count = VALUES(order_count),
total_amount = VALUES(total_amount);
应用层缓存
class AggregationCache {
private $redis;
private $cacheTTL = 300; // 5分钟
public function getCachedAggregation($cacheKey, $callback) {
// 检查缓存
$cached = $this->redis->get($cacheKey);
if ($cached !== false) {
return json_decode($cached, true);
}
// 执行聚合查询
$result = $callback();
// 缓存结果
$this->redis->setex($cacheKey, $this->cacheTTL, json_encode($result));
return $result;
}
}
索引优化策略
-- 在每个分片创建覆盖索引 CREATE INDEX idx_status_created ON orders (status, created_at) INCLUDE (amount); -- 对于特定查询的高效索引 CREATE INDEX idx_user_month ON orders (user_id, created_at) INCLUDE (amount, status);
分片键优化
选择合适的分片键
class ShardKeyOptimizer {
// 根据查询模式选择分片键
public function determineShardKey($queryPattern) {
switch ($queryPattern) {
case 'user_centric':
return 'user_id';
case 'time_range':
return 'date_range';
case 'mixed':
return 'hash(user_id, date_range)';
}
}
// 复合分片键设计
public function getShardKey($userId, $date) {
// 根据用户ID和时间范围决定分片
$month = date('Ym', strtotime($date));
return crc32($userId . '_' . $month) % $this->shardCount;
}
}
查询优化示例
分页聚合查询
class PagedAggregationQuery {
public function getPagedAggregation($page, $pageSize) {
$allResults = [];
$offset = ($page - 1) * $pageSize;
// 在应用层进行跨分片排序和分页
foreach ($this->shards as $shard) {
$results = $shard->query(
"SELECT * FROM (
SELECT
user_id,
COUNT(*) as order_count,
SUM(amount) as total_amount
FROM orders
GROUP BY user_id
ORDER BY total_amount DESC
LIMIT $offset, $pageSize
) as t"
);
$allResults = array_merge($allResults, $results);
}
// 应用层排序和分页
usort($allResults, function($a, $b) {
return $b['total_amount'] <=> $a['total_amount'];
});
return array_slice($allResults, 0, $pageSize);
}
}
监控与调优
class AggregationMonitor {
public function recordQueryPerformance($queryId, $startTime, $endTime) {
$duration = $endTime - $startTime;
// 记录到监控系统
$this->metricsCollector->record([
'query_id' => $queryId,
'duration' => $duration,
'shards_queried' => count($this->shards),
'timestamp' => time()
]);
// 慢查询告警
if ($duration > 2.0) {
$this->alertSystem->sendAlert("Slow aggregation query: $queryId");
}
}
}
实际应用示例
class OrderStatisticsService {
private $shardQueryOptimizer;
private $cacheService;
public function getMonthlyStatistics($year, $month) {
$cacheKey = "statistics:monthly:{$year}-{$month}";
return $this->cacheService->get($cacheKey, function() use ($year, $month) {
$startDate = "{$year}-{$month}-01";
$endDate = date('Y-m-t', strtotime($startDate));
// 并行查询所有分片
$partialResults = $this->shardQueryOptimizer->parallelQuery(
$this->buildShardQueries($startDate, $endDate)
);
// 合并结果
return $this->mergeResults($partialResults);
}, 600); // 缓存10分钟
}
private function buildShardQueries($startDate, $endDate) {
$queries = [];
foreach ($this->shards as $shardName => $shard) {
$queries[$shardName] = "
SELECT
DATE(created_at) as date,
COUNT(DISTINCT user_id) as unique_users,
COUNT(*) as total_orders,
SUM(amount) as total_revenue,
AVG(amount) as avg_order_amount
FROM orders
WHERE created_at BETWEEN '$startDate' AND '$endDate'
GROUP BY DATE(created_at)
";
}
return $queries;
}
}
这些优化策略可以显著改善跨分片聚合查询的性能，建议根据具体的业务场景和数据特点，选择合适的策略组合，定期监控查询性能,及时调整优化策略。
