PHPAPI任务分片怎么实现

wen PHP项目 2026-07-05 211

本文目录导读：

PHPAPI任务分片怎么实现

基础分片实现（基于ID取模）
基于范围的分片（Range-based）
分布式任务分片（Redis实现）
高级：一致性哈希分片
完整生产级示例
实际应用建议

在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取模适合均匀分布的数据，一致性哈希适合节点动态变化的场景,范围分片适合有顺序要求的数据处理。

本文地址： https://www.bmjup.com/post/16635.html