PHPAPI任务去重怎么实现

本文目录导读：

1. 基于幂等性设计（推荐）
2. 基于数据库的唯一约束
3. 基于消息队列去重
4. 基于时间窗口去重
5. 综合实现示例
6. 客户端实现建议
7. 推荐方案选择
8. 最佳实践建议

在PHP API开发中，任务去重是一个常见且重要的问题,下面我介绍几种常用的实现方案：

基于幂等性设计（推荐）

唯一请求ID（幂等键）

class IdempotentApi
{
    private $redis;
    public function __construct($redis) 
    {
        $this->redis = $redis;
    }
    /**
     * 处理幂等请求
     * @param string $idempotentKey 客户端生成的唯一键
     * @param callable $callback 实际业务处理逻辑
     * @param int $ttl 过期时间（秒）
     */
    public function handle(string $idempotentKey, callable $callback, int $ttl = 3600)
    {
        // 检查是否已处理
        $result = $this->redis->get("idempotent:{$idempotentKey}");
        if ($result !== null) {
            // 返回之前的结果
            return unserialize($result);
        }
        // 使用SET NX实现原子操作
        $lockKey = "lock:{$idempotentKey}";
        $lockAcquired = $this->redis->set(
            $lockKey, 
            'processing', 
            ['NX', 'EX' => 30] // 30秒锁超时
        );
        if (!$lockAcquired) {
            throw new \RuntimeException('请求正在处理中');
        }
        try {
            // 执行实际业务逻辑
            $result = $callback();
            // 缓存结果
            $this->redis->setex(
                "idempotent:{$idempotentKey}",
                $ttl,
                serialize($result)
            );
            return $result;
        } finally {
            // 释放锁
            $this->redis->del($lockKey);
        }
    }
}
// 使用示例
$api = new IdempotentApi($redis);
// 客户端需要生成唯一的idempotent_key
$idempotentKey = $_SERVER['HTTP_X_IDEMPOTENT_KEY'] ?? uniqid();
$result = $api->handle($idempotentKey, function() {
    // 实际的业务处理
    return createOrder($data);
});

基于数据库的唯一约束

MySQL实现

class DuplicateCheckApi 
{
    private $pdo;
    /**
     * 数据库表结构：
     * CREATE TABLE task_queue (
     *     id INT AUTO_INCREMENT PRIMARY KEY,
     *     task_key VARCHAR(64) NOT NULL UNIQUE,
     *     task_data JSON,
     *     status ENUM('pending', 'processing', 'completed', 'failed') DEFAULT 'pending',
     *     result JSON,
     *     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
     *     updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
     * );
     */
    public function submitTask(string $taskKey, array $taskData)
    {
        try {
            // 使用INSERT IGNORE或ON DUPLICATE KEY
            $stmt = $this->pdo->prepare(
                'INSERT INTO task_queue (task_key, task_data, status) 
                 VALUES (?, ?, \'pending\')
                 ON DUPLICATE KEY UPDATE id=id'
            );
            $stmt->execute([$taskKey, json_encode($taskData)]);
            if ($stmt->rowCount() === 0) {
                // 任务已存在，返回已有任务信息
                return $this->getTaskStatus($taskKey);
            }
            // 新任务，处理业务逻辑
            return $this->processTask($taskKey);
        } catch (\PDOException $e) {
            // 处理并发冲突
            if ($e->getCode() == 23000) { // 重复键异常
                return $this->getTaskStatus($taskKey);
            }
            throw $e;
        }
    }
}

基于消息队列去重

RabbitMQ + Redis去重

class MessageDeduplicator
{
    private $redis;
    private $queue;
    public function publishDeduplicated(string $messageId, array $messageData)
    {
        // 使用HyperLogLog或Set进行去重
        $isDuplicate = $this->redis->sadd('msg_set', $messageId);
        if ($isDuplicate === false) {
            // 已存在的消息
            return false;
        }
        // 设置过期时间，避免内存溢出
        $this->redis->expire('msg_set', 86400); // 24小时
        // 发布到消息队列
        $this->queue->publish(json_encode([
            'id' => $messageId,
            'data' => $messageData,
            'timestamp' => time()
        ]));
        return true;
    }
}

基于时间窗口去重

class TimeWindowDeduplicator
{
    private $redis;
    private $windowSize = 60; // 60秒窗口
    /**
     * 基于滑动时间窗口去重
     */
    public function isDuplicate(string $taskId): bool
    {
        $key = "task_window:{$taskId}";
        $now = time();
        // 使用有序集合，时间戳作为score
        $this->redis->zadd($key, $now, "{$now}:{$taskId}");
        // 移除窗口外的数据
        $this->redis->zremrangebyscore($key, 0, $now - $this->windowSize);
        // 检查窗口内是否有重复
        $count = $this->redis->zcard($key);
        return $count > 1;
    }
}

综合实现示例

class TaskDeduplicationService
{
    private $redis;
    private $db;
    private $lockTimeout = 30;
    private $cacheTimeout = 3600;
    /**
     * 完整的去重处理流程
     */
    public function processTask(string $taskId, array $taskData, callable $handler)
    {
        // 1. 幂等性检查
        $cacheKey = "task_result:{$taskId}";
        $cachedResult = $this->redis->get($cacheKey);
        if ($cachedResult !== null) {
            return unserialize($cachedResult);
        }
        // 2. 分布式锁 - 防止并发
        $lockKey = "task_lock:{$taskId}";
        $lockToken = uniqid('', true);
        $locked = $this->redis->set(
            $lockKey,
            $lockToken,
            ['NX', 'EX' => $this->lockTimeout]
        );
        if (!$locked) {
            // 等待重试或返回冲突
            return $this->handleConcurrency($taskId);
        }
        try {
            // 3. 数据库幂等检查
            if ($this->isTaskProcessed($taskId)) {
                return $this->getTaskResult($taskId);
            }
            // 4. 执行任务
            $result = $handler($taskData);
            // 5. 标记任务完成
            $this->markTaskCompleted($taskId, $result);
            // 6. 缓存结果
            $this->redis->setex(
                $cacheKey,
                $this->cacheTimeout,
                serialize($result)
            );
            return $result;
        } catch (\Exception $e) {
            $this->handleError($taskId, $e);
            throw $e;
        } finally {
            // 7. 释放锁（使用Lua脚本保证原子性）
            $this->releaseLock($lockKey, $lockToken);
        }
    }
    private function releaseLock(string $key, string $token)
    {
        $script = <<<LUA
if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end
LUA;
        $this->redis->eval($script, [$key, $token], 1);
    }
    private function handleConcurrency(string $taskId)
    {
        // 重试逻辑
        $maxRetries = 3;
        $retryDelay = 100; // 毫秒
        for ($i = 0; $i < $maxRetries; $i++) {
            usleep($retryDelay * 1000);
            $cachedResult = $this->redis->get("task_result:{$taskId}");
            if ($cachedResult !== null) {
                return unserialize($cachedResult);
            }
        }
        throw new \RuntimeException("任务{$taskId}正在处理中，请稍后重试");
    }
}

客户端实现建议

// 客户端生成唯一ID
class ClientIdGenerator
{
    public static function generateRequestId(): string
    {
        // UUID v4
        $data = random_bytes(16);
        $data[6] = chr(ord($data[6]) & 0x0f | 0x40);
        $data[8] = chr(ord($data[8]) & 0x3f | 0x80);
        return vsprintf('%s%s-%s-%s-%s-%s%s%s', str_split(bin2hex($data), 4));
    }
}
// 客户端调用示例
$client = new GuzzleHttp\Client();
$requestId = ClientIdGenerator::generateRequestId();
$response = $client->post('/api/orders', [
    'json' => $data,
    'headers' => [
        'X-Idempotent-Key' => $requestId,
        'X-Request-Id' => $requestId
    ]
]);

推荐方案选择

最佳实践建议

1. 客户端生成唯一ID，服务端不做ID生成
2. 使用分布式锁防止并发问题
3. 设置合理的TTL，避免存储无限增长
4. 考虑幂等性和去重的关系，幂等允许重复请求但结果一致
5. 做好日志记录，方便问题排查
6. 返回清晰的状态码，如409 Conflict表示重复请求

选择哪种方案取决于你的具体业务场景、并发量和系统架构。