本文目录导读：
在PHP API中实现任务分片（Task Sharding/Slicing）通常有几种主流方案,我为你详细讲解最实用的实现方式：
基础分片实现（基于ID取模）
<?php
class TaskSharder {
private $totalShards; // 总分片数
private $shardKey; // 分片键
public function __construct($totalShards = 10) {
$this->totalShards = $totalShards;
}
// 根据ID计算所属分片
public function getShardById($id) {
return crc32((string)$id) % $this->totalShards;
}
// 批量任务分片
public function shardTasks(array $tasks) {
$sharded = array_fill(0, $this->totalShards, []);
foreach ($tasks as $task) {
$shardId = $this->getShardById($task['id']);
$sharded[$shardId][] = $task;
}
return array_filter($sharded);
}
}
// 使用示例
$sharder = new TaskSharder(5);
$tasks = [
['id' => 1, 'data' => 'task1'],
['id' => 2, 'data' => 'task2'],
// ... 更多任务
];
$shardedTasks = $sharder->shardTasks($tasks);
基于范围的分片（Range-based）
<?php
class RangeTaskSharder {
private $maxId;
private $shardCount;
private $ranges;
public function __construct($maxId, $shardCount = 10) {
$this->maxId = $maxId;
$this->shardCount = $shardCount;
$this->calculateRanges();
}
private function calculateRanges() {
$chunkSize = ceil($this->maxId / $this->shardCount);
for ($i = 0; $i < $this->shardCount; $i++) {
$this->ranges[] = [
'start' => $i * $chunkSize,
'end' => min(($i + 1) * $chunkSize - 1, $this->maxId)
];
}
}
public function getShardRange($shardId) {
return $this->ranges[$shardId] ?? null;
}
public function getTasksForShard($shardId, $db) {
$range = $this->getShardRange($shardId);
if (!$range) return [];
$sql = "SELECT * FROM tasks
WHERE id BETWEEN ? AND ?";
return $db->fetchAll($sql, [$range['start'], $range['end']]);
}
}
分布式任务分片（Redis实现）
<?php
class RedisTaskSharder {
private $redis;
private $shardPrefix = 'task_shard:';
private $taskPrefix = 'task:';
public function __construct(Redis $redis) {
$this->redis = $redis;
}
// 添加任务到分片
public function addTask($taskId, $taskData, $shardCount = 10) {
$shardId = crc32($taskId) % $shardCount;
$shardKey = $this->shardPrefix . $shardId;
$taskKey = $this->taskPrefix . $taskId;
// 使用管道提高性能
$this->redis->multi()
->lPush($shardKey, $taskId)
->hMSet($taskKey, $taskData)
->exec();
return $shardId;
}
// 从指定分片获取任务
public function getTasksFromShard($shardId, $count = 10) {
$shardKey = $this->shardPrefix . $shardId;
$taskIds = $this->redis->lRange($shardKey, 0, $count - 1);
$tasks = [];
foreach ($taskIds as $taskId) {
$taskKey = $this->taskPrefix . $taskId;
$tasks[] = $this->redis->hGetAll($taskKey);
}
return $tasks;
}
// 消费任务
public function consumeTaskFromShard($shardId) {
$shardKey = $this->shardPrefix . $shardId;
$taskId = $this->redis->rPop($shardKey);
if ($taskId) {
$taskKey = $this->taskPrefix . $taskId;
$taskData = $this->redis->hGetAll($taskKey);
$this->redis->del($taskKey);
return $taskData;
}
return null;
}
}
高级：一致性哈希分片
<?php
class ConsistentHashSharder {
private $nodes = [];
private $virtualNodes = [];
private $totalVirtualNodes = 64;
public function addNode($node, $weight = 1) {
for ($i = 0; $i < $this->totalVirtualNodes * $weight; $i++) {
$hash = crc32("{$node}:{$i}");
$this->virtualNodes[$hash] = $node;
}
ksort($this->virtualNodes);
$this->nodes[] = $node;
}
public function getNode($key) {
if (empty($this->virtualNodes)) {
return null;
}
$hash = crc32((string)$key);
// 找到最近的虚拟节点
foreach ($this->virtualNodes as $virtualHash => $node) {
if ($virtualHash >= $hash) {
return $node;
}
}
// 如果没找到，返回第一个节点
return reset($this->virtualNodes);
}
public function removeNode($node) {
foreach ($this->virtualNodes as $hash => $n) {
if ($n === $node) {
unset($this->virtualNodes[$hash]);
}
}
$this->nodes = array_diff($this->nodes, [$node]);
}
}
完整生产级示例
<?php
class TaskProcessingSystem {
private $db;
private $redis;
private $shardCount;
public function __construct($shardCount = 10) {
$this->shardCount = $shardCount;
$this->db = Database::getInstance();
$this->redis = Redis::getInstance();
}
// 分片处理大数据集
public function processLargeDataset($table, $callback) {
$totalRecords = $this->db->fetchColumn("SELECT COUNT(*) FROM {$table}");
$batchSize = ceil($totalRecords / $this->shardCount);
for ($i = 0; $i < $this->shardCount; $i++) {
$offset = $i * $batchSize;
$tasks[] = [
'shard_id' => $i,
'offset' => $offset,
'limit' => $batchSize,
'table' => $table
];
}
// 并行处理每个分片
return $this->parallelProcess($tasks, $callback);
}
private function parallelProcess($tasks, $callback) {
$results = [];
$workers = [];
// 使用pcntl_fork进行多进程处理
foreach ($tasks as $shardId => $task) {
$pid = pcntl_fork();
if ($pid == -1) {
die("Fork failed");
} else if ($pid) {
// 父进程
$workers[] = $pid;
} else {
// 子进程
$result = $callback($task);
exit($result ? 0 : 1);
}
}
// 等待所有子进程完成
foreach ($workers as $worker) {
pcntl_waitpid($worker, $status);
$results[] = pcntl_wexitstatus($status);
}
return $results;
}
// 动态调整分片
public function rebalanceShards($newShardCount) {
$this->shardCount = $newShardCount;
// 重新分配任务
// 实现重平衡逻辑
}
}
实际应用建议
选择合适的策略：
// 1. 对于均匀分布的数据 $sharder = new TaskSharder(10); // 简单取模 // 2. 对于有范围查询需求 $sharder = new RangeTaskSharder(1000000, 10); // 范围分片 // 3. 对于分布式系统 $sharder = new ConsistentHashSharder(); // 一致性哈希
性能优化技巧：
// 1. 预计算分片
$shardCache = [];
function getCachedShard($id) {
global $shardCache;
if (!isset($shardCache[$id])) {
$shardCache[$id] = crc32($id) % 10;
}
return $shardCache[$id];
}
// 2. 批量操作
$redis->pipeline(function($pipe) use ($shardedTasks) {
foreach ($shardedTasks as $shardId => $tasks) {
$pipe->lPush("shard:{$shardId}", ...$tasks);
}
});
监控和运维：
// 监控每个分片的负载
class ShardMonitor {
public function getShardMetrics() {
$metrics = [];
for ($i = 0; $i < 10; $i++) {
$metrics[$i] = [
'queue_size' => $this->redis->lLen("shard:{$i}"),
'processing_time' => $this->getAvgProcessingTime($i),
'error_rate' => $this->getErrorRate($i)
];
}
return $metrics;
}
}
选择哪种方案取决于你的具体需求：简单的ID取模适合均匀分布的数据，一致性哈希适合节点动态变化的场景,范围分片适合有顺序要求的数据处理。
