PHP任务失败重试设计

本文目录导读：

1. 基础重试机制设计
2. 持久化任务队列
3. 完整的任务处理器
4. 独立重试工具类
5. 使用示例
6. 数据库表结构
7. 最佳实践建议

我来设计一个完善的PHP任务失败重试系统。

基础重试机制设计

重试策略配置类

<?php
namespace App\Retry;
/**
 * 重试策略配置
 */
class RetryPolicy
{
    /**
     * @var int 最大重试次数
     */
    private int $maxAttempts = 3;
    /**
     * @var array 重试延迟（秒），基于指数退避
     */
    private array $delaySeconds = [1, 5, 15, 30, 60];
    /**
     * @var array 需要重试的异常类列表
     */
    private array $retryableExceptions = [];
    /**
     * @var callable|null 自定义重试条件
     */
    private $retryConditionCallback = null;
    /**
     * @var bool 是否启用抖动
     */
    private bool $enableJitter = true;
    /**
     * @var bool 是否记录重试日志
     */
    private bool $enableLogging = true;
    // Getter和Setter方法...
    public function getMaxAttempts(): int
    {
        return $this->maxAttempts;
    }
    public function setMaxAttempts(int $maxAttempts): self
    {
        $this->maxAttempts = $maxAttempts;
        return $this;
    }
    public function getDelaySeconds(): array
    {
        return $this->delaySeconds;
    }
    public function setDelaySeconds(array $delaySeconds): self
    {
        $this->delaySeconds = $delaySeconds;
        return $this;
    }
    public function getRetryableExceptions(): array
    {
        return $this->retryableExceptions;
    }
    public function setRetryableExceptions(array $exceptions): self
    {
        $this->retryableExceptions = $exceptions;
        return $this;
    }
    public function getRetryConditionCallback(): ?callable
    {
        return $this->retryConditionCallback;
    }
    public function setRetryConditionCallback(callable $callback): self
    {
        $this->retryConditionCallback = $callback;
        return $this;
    }
    public function isEnableJitter(): bool
    {
        return $this->enableJitter;
    }
    public function setEnableJitter(bool $enableJitter): self
    {
        $this->enableJitter = $enableJitter;
        return $this;
    }
    public function isEnableLogging(): bool
    {
        return $this->enableLogging;
    }
    public function setEnableLogging(bool $enableLogging): self
    {
        $this->enableLogging = $enableLogging;
        return $this;
    }
    /**
     * 获取指定尝试次数的延迟时间（带抖动）
     */
    public function getDelayForAttempt(int $attempt): int
    {
        $index = min($attempt, count($this->delaySeconds) - 1);
        $delay = $this->delaySeconds[$index];
        if ($this->enableJitter) {
            // 添加±20%的随机抖动
            $jitter = $delay * 0.2 * (mt_rand() / mt_getrandmax() * 2 - 1);
            $delay = max(0, $delay + $jitter);
        }
        return (int) $delay;
    }
}

重试执行器

<?php
namespace App\Retry;
use Psr\Log\LoggerInterface;
use Throwable;
/**
 * 重试执行器
 */
class RetryExecutor
{
    private LoggerInterface $logger;
    private array $retryListeners = [];
    public function __construct(?LoggerInterface $logger = null)
    {
        $this->logger = $logger;
    }
    /**
     * 执行任务并支持重试
     *
     * @param callable $task 要执行的任务
     * @param RetryPolicy|null $policy 重试策略
     * @return mixed 任务结果
     * @throws Throwable 当所有重试都失败时抛出最后的异常
     */
    public function execute(callable $task, ?RetryPolicy $policy = null): mixed
    {
        $policy = $policy ?? new RetryPolicy();
        $attempt = 1;
        $lastException = null;
        while ($attempt <= $policy->getMaxAttempts()) {
            try {
                // 执行任务前的回调
                $this->triggerEvent('beforeRetry', $attempt, $policy);
                // 执行任务
                $result = $task();
                // 执行成功
                $this->triggerEvent('onSuccess', $attempt, $policy);
                return $result;
            } catch (Throwable $e) {
                $lastException = $e;
                // 检查是否应该重试
                if (!$this->shouldRetry($e, $attempt, $policy)) {
                    $this->triggerEvent('onFailure', $attempt, $policy, $e);
                    throw $e;
                }
                // 计算延迟时间
                $delay = $policy->getDelayForAttempt($attempt - 1);
                // 记录日志
                if ($policy->isEnableLogging() && $this->logger) {
                    $this->logger->warning(
                        "Task failed, retrying in {$delay}s. Attempt {$attempt}/{$policy->getMaxAttempts()}",
                        [
                            'exception' => $e->getMessage(),
                            'trace' => $e->getTraceAsString(),
                        ]
                    );
                }
                // 触发重试事件
                $this->triggerEvent('onRetry', $attempt, $policy, $e);
                // 等待后重试
                if ($delay > 0) {
                    sleep($delay);
                }
                $attempt++;
            }
        }
        // 所有重试都失败了
        $this->triggerEvent('onFailure', $attempt - 1, $policy, $lastException);
        throw $lastException;
    }
    /**
     * 判断是否应该重试
     */
    private function shouldRetry(Throwable $e, int $attempt, RetryPolicy $policy): bool
    {
        // 超过最大次数不重试
        if ($attempt >= $policy->getMaxAttempts()) {
            return false;
        }
        // 自定义重试条件
        $callback = $policy->getRetryConditionCallback();
        if ($callback !== null) {
            return (bool) call_user_func($callback, $e, $attempt);
        }
        // 检查异常类型配置
        $retryableExceptions = $policy->getRetryableExceptions();
        if (!empty($retryableExceptions)) {
            foreach ($retryableExceptions as $exceptionClass) {
                if ($e instanceof $exceptionClass) {
                    return true;
                }
            }
            return false;
        }
        // 默认所有异常都重试
        return true;
    }
    /**
     * 注册重试监听器
     */
    public function addListener(string $event, callable $callback): void
    {
        $this->retryListeners[$event][] = $callback;
    }
    /**
     * 触发事件
     */
    private function triggerEvent(string $event, int $attempt, RetryPolicy $policy, ?Throwable $e = null): void
    {
        if (!isset($this->retryListeners[$event])) {
            return;
        }
        foreach ($this->retryListeners[$event] as $callback) {
            call_user_func($callback, $attempt, $policy, $e);
        }
    }
}

持久化任务队列

任务队列类

<?php
namespace App\Queue;
use PDO;
use Redis;
/**
 * 持久化任务队列
 */
class TaskQueue
{
    private PDO $db;
    private ?Redis $redis = null;
    private string $table;
    public function __construct(PDO $db, string $table = 'tasks', ?Redis $redis = null)
    {
        $this->db = $db;
        $this->table = $table;
        $this->redis = $redis;
    }
    /**
     * 创建任务
     */
    public function enqueue(string $taskType, array $payload, int $maxRetries = 3): int
    {
        $sql = "INSERT INTO {$this->table} 
                (task_type, payload, status, attempts, max_retries, created_at, updated_at) 
                VALUES (?, ?, 'pending', 0, ?, NOW(), NOW())";
        $stmt = $this->db->prepare($sql);
        $stmt->execute([
            $taskType,
            json_encode($payload),
            $maxRetries
        ]);
        $taskId = (int) $this->db->lastInsertId();
        // 如果有Redis，推送到队列
        if ($this->redis) {
            $this->redis->lPush('task_queue', $taskId);
        }
        return $taskId;
    }
    /**
     * 获取待处理任务
     */
    public function claimTask(int $workerId): ?array
    {
        $sql = "SELECT * FROM {$this->table} 
                WHERE status = 'pending' OR 
                      (status = 'failed' AND attempts < max_retries) 
                ORDER BY created_at ASC 
                LIMIT 1 FOR UPDATE SKIP LOCKED";
        $stmt = $this->db->prepare($sql);
        $stmt->execute();
        $task = $stmt->fetch(PDO::FETCH_ASSOC);
        if ($task) {
            // 标记为执行中
            $updateSql = "UPDATE {$this->table} 
                         SET status = 'processing', 
                             worker_id = ?,
                             started_at = NOW(),
                             updated_at = NOW() 
                         WHERE id = ?";
            $updateStmt = $this->db->prepare($updateSql);
            $updateStmt->execute([$workerId, $task['id']]);
            return $task;
        }
        return null;
    }
    /**
     * 标记任务成功
     */
    public function markSuccess(int $taskId, array $result = []): void
    {
        $sql = "UPDATE {$this->table} 
                SET status = 'success', 
                    result = ?,
                    completed_at = NOW(),
                    updated_at = NOW() 
                WHERE id = ?";
        $stmt = $this->db->prepare($sql);
        $stmt->execute([json_encode($result), $taskId]);
    }
    /**
     * 标记任务失败（准备重试）
     */
    public function markFailed(int $taskId, string $error): bool
    {
        // 检查任务信息
        $task = $this->getTask($taskId);
        if (!$task || $task['attempts'] >= $task['max_retries']) {
            $this->markDead($taskId, $error);
            return false;
        }
        $attempts = $task['attempts'] + 1;
        $sql = "UPDATE {$this->table} 
                SET status = 'failed',
                    attempts = ?,
                    last_error = ?,
                    next_attempt_at = DATE_ADD(NOW(), INTERVAL ? SECOND),
                    updated_at = NOW() 
                WHERE id = ?";
        $stmt = $this->db->prepare($sql);
        // 计算延迟（指数退避）
        $delay = min(300, pow(2, $attempts) * 5); // 5s, 10s, 20s...
        $stmt->execute([$attempts, $error, $delay, $taskId]);
        return true;
    }
    /**
     * 标记死信任务
     */
    public function markDead(int $taskId, string $error): void
    {
        $sql = "UPDATE {$this->table} 
                SET status = 'dead',
                    last_error = ?,
                    completed_at = NOW(),
                    updated_at = NOW() 
                WHERE id = ?";
        $stmt = $this->db->prepare($sql);
        $stmt->execute([$error, $taskId]);
    }
    /**
     * 获取任务
     */
    public function getTask(int $taskId): ?array
    {
        $sql = "SELECT * FROM {$this->table} WHERE id = ?";
        $stmt = $this->db->prepare($sql);
        $stmt->execute([$taskId]);
        $task = $stmt->fetch(PDO::FETCH_ASSOC);
        return $task ?: null;
    }
    /**
     * 获取统计信息
     */
    public function getStats(): array
    {
        $sql = "SELECT status, COUNT(*) as count 
                FROM {$this->table} 
                GROUP BY status";
        $stmt = $this->db->query($sql);
        $stats = [];
        while ($row = $stmt->fetch(PDO::FETCH_ASSOC)) {
            $stats[$row['status']] = (int) $row['count'];
        }
        return $stats;
    }
}

完整的任务处理器

<?php
namespace App\Queue;
use App\Retry\RetryExecutor;
use App\Retry\RetryPolicy;
use Psr\Log\LoggerInterface;
/**
 * 任务处理器基类
 */
abstract class TaskHandler
{
    protected TaskQueue $queue;
    protected RetryExecutor $retryExecutor;
    protected LoggerInterface $logger;
    public function __construct(TaskQueue $queue, ?LoggerInterface $logger = null)
    {
        $this->queue = $queue;
        $this->logger = $logger ?? new NullLogger();
        $this->retryExecutor = new RetryExecutor($logger);
    }
    /**
     * 执行任务
     */
    public function handle(array $task): void
    {
        $taskId = (int) $task['id'];
        $payload = json_decode($task['payload'], true);
        try {
            // 验证任务数据
            if (!$this->validate($payload)) {
                $this->queue->markFailed($taskId, 'Validation failed');
                return;
            }
            // 执行任务
            $result = $this->process($payload);
            // 标记成功
            $this->queue->markSuccess($taskId, $result);
        } catch (\Exception $e) {
            // 记录日志
            $this->logger->error("Task {$taskId} failed: " . $e->getMessage());
            // 标记失败，决定是否重试
            if (!$this->queue->markFailed($taskId, $e->getMessage())) {
                // 已达到最大重试次数
                $this->onDeadLetter($taskId, $payload, $e);
            }
        }
    }
    /**
     * 处理任务
     */
    abstract protected function process(array $payload): mixed;
    /**
     * 验证任务数据
     */
    protected function validate(array $payload): bool
    {
        return !empty($payload);
    }
    /**
     * 死信处理（可重写）
     */
    protected function onDeadLetter(int $taskId, array $payload, \Exception $e): void
    {
        $this->logger->error("Task {$taskId} moved to dead letter queue");
    }
}
/**
 * 示例任务
 */
class OrderTaskHandler extends TaskHandler
{
    /**
     * 示例：处理订单任务
     */
    protected function process(array $payload): mixed
    {
        // 模拟处理业务
        $orderId = $payload['order_id'];
        $action = $payload['action'] ?? 'create';
        switch ($action) {
            case 'create':
                return $this->createOrder($orderId);
            case 'update':
                return $this->updateOrder($orderId, $payload['data'] ?? []);
            default:
                throw new \InvalidArgumentException("Unknown action: {$action}");
        }
    }
    private function createOrder(int $orderId): array
    {
        // 模拟业务逻辑
        if ($orderId % 5 == 0) {
            throw new \RuntimeException("Database connection failed");
        }
        return ['status' => 'created', 'order_id' => $orderId];
    }
    private function updateOrder(int $orderId, array $data): array
    {
        return ['status' => 'updated', 'order_id' => $orderId, 'data' => $data];
    }
}

独立重试工具类

<?php
namespace App\Retry;
use Closure;
use Throwable;
/**
 * 独立的重试工具类
 */
class Retry
{
    /**
     * 带重试的闭包执行
     *
     * @param Closure $fn 需要重试的函数
     * @param array $options 配置选项
     * @return mixed
     */
    public static function withRetry(Closure $fn, array $options = []): mixed
    {
        $attempts = $options['attempts'] ?? 3;
        $delay = $options['delay'] ?? 1; // 基础延迟秒数
        $multiplier = $options['multiplier'] ?? 2; // 延迟倍数
        $maxDelay = $options['max_delay'] ?? 60; // 最大延迟
        $jitter = $options['jitter'] ?? false; // 是否添加抖动
        $onRetry = $options['on_retry'] ?? null; // 重试回调
        $retryCondition = $options['retry_condition'] ?? null; // 重试条件
        $attempt = 1;
        $lastException = null;
        while (true) {
            try {
                return $fn();
            } catch (Throwable $e) {
                $lastException = $e;
                // 检查是否需要重试
                if ($attempt >= $attempts) {
                    break;
                }
                // 检查重试条件（如果有）
                if ($retryCondition && !call_user_func($retryCondition, $e)) {
                    break;
                }
                // 计算延迟时间
                $waitTime = min($maxDelay, $delay * pow($multiplier, $attempt - 1));
                if ($jitter) {
                    // 添加随机抖动（±20%）
                    $waitTime *= (1 + (mt_rand() / mt_getrandmax() - 0.5) * 0.2);
                }
                // 重试回调
                if ($onRetry) {
                    call_user_func($onRetry, $attempt, $waitTime, $e);
                }
                sleep(max(0, (int) $waitTime));
                $attempt++;
            }
        }
        throw $lastException;
    }
    /**
     * 异步重试辅助类
     */
    public static function async(callable $fn, array $options = []): mixed
    {
        // 示例：在异步环境中使用
        return self::withRetry($fn, $options);
    }
    /**
     * 并发重试多个任务
     */
    public static function all(array $tasks, array $options = []): array
    {
        $results = [];
        $errors = [];
        foreach ($tasks as $key => $task) {
            try {
                $results[$key] = self::withRetry($task, $options);
            } catch (Throwable $e) {
                $errors[$key] = $e;
            }
        }
        if (!empty($errors)) {
            throw new \RuntimeException(
                'Multiple tasks failed: ' . implode(', ', array_keys($errors))
            );
        }
        return $results;
    }
}

使用示例

<?php
// 1. 使用基础重试类
use App\Retry\RetryExecutor;
use App\Retry\RetryPolicy;
$retryPolicy = new RetryPolicy();
$retryPolicy->setMaxAttempts(3)
    ->setDelaySeconds([1, 5, 15])
    ->setRetryableExceptions([
        \PDOException::class,
        \RedisException::class,
    ])
    ->setEnableJitter(true)
    ->setEnableLogging(true);
$executor = new RetryExecutor($logger);
try {
    $result = $executor->execute(function () {
        // 你的业务逻辑
        return performBusinessOperation();
    }, $retryPolicy);
} catch (\Exception $e) {
    echo "任务最终失败: " . $e->getMessage();
}
// 2. 使用简洁重试方法
use App\Retry\Retry;
$result = Retry::withRetry(function () {
    // 业务代码
    return $this->callExternalApi();
}, [
    'attempts' => 5,
    'delay' => 1,
    'multiplier' => 2,
    'max_delay' => 30,
    'jitter' => true,
    'retry_condition' => function ($exception) {
        // 只重试特定异常
        return $exception instanceof \RuntimeException;
    },
    'on_retry' => function ($attempt, $delay, $exception) {
        Log::warning("第{$attempt}次重试，等待{$delay}秒", [
            'error' => $exception->getMessage()
        ]);
    }
]);
// 3. 使用任务队列
use App\Queue\TaskQueue;
$queue = new TaskQueue($pdo, 'tasks', $redis);
// 添加任务
$taskId = $queue->enqueue(
    'order.process',
    ['order_id' => 123],
    3 // 最大重试3次
);
// 处理任务（worker进程）
$task = $queue->claimTask($workerId);
// 根据任务类型分发处理
$handlerMap = [
    'order.process' => new OrderTaskHandler($queue, $logger),
    'email.send' => new EmailTaskHandler($queue, $logger),
];
if ($task && isset($handlerMap[$task['task_type']])) {
    $handler = $handlerMap[$task['task_type']];
    $handler->handle($task);
}

数据库表结构

-- 任务表
CREATE TABLE IF NOT EXISTS `tasks` (
  `id` BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
  `task_type` VARCHAR(100) NOT NULL COMMENT '任务类型',
  `payload` JSON NOT NULL COMMENT '任务数据',
  `status` ENUM('pending', 'processing', 'success', 'failed', 'dead') 
           DEFAULT 'pending' COMMENT '任务状态',
  `attempts` INT UNSIGNED DEFAULT 0 COMMENT '已尝试次数',
  `max_retries` INT UNSIGNED DEFAULT 3 COMMENT '最大重试次数',
  `worker_id` VARCHAR(64) DEFAULT NULL COMMENT '处理worker',
  `result` JSON DEFAULT NULL COMMENT '成功结果',
  `last_error` TEXT DEFAULT NULL COMMENT '最后错误信息',
  `next_attempt_at` DATETIME DEFAULT NULL COMMENT '下次重试时间',
  `started_at` DATETIME DEFAULT NULL COMMENT '开始时间',
  `completed_at` DATETIME DEFAULT NULL COMMENT '完成时间',
  `created_at` DATETIME NOT NULL,
  `updated_at` DATETIME NOT NULL,
  INDEX `idx_status_created` (`status`, `created_at`),
  INDEX `idx_next_attempt` (`status`, `next_attempt_at`),
  INDEX `idx_worker` (`worker_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

最佳实践建议

错误分类处理

class RetryableException extends \Exception {}
class NonRetryableException extends \Exception {}
// 使用方式
if ($e instanceof NonRetryableException) {
    // 不重试，直接失败
} else {
    // 可以进行重试
}

监控和告警

// 添加监控指标
class RetryMonitor
{
    public static function recordRetry(string $taskType, int $attempt): void
    {
        // 发送到监控系统（如Prometheus）
        Metrics::increment("task.retry", [
            'type' => $taskType,
            'attempt' => $attempt
        ]);
    }
    public static function recordDeadLetter(string $taskType): void
    {
        Metrics::increment("task.dead_letter", [
            'type' => $taskType
        ]);
    }
}

这个设计提供了完整的任务重试机制,包括：

1. 灵活的配置：支持自定义重试次数、延迟策略、重试条件
2. 持久化存储：数据库存储任务状态，支持分布式处理
3. 指数退避：智能延迟算法避免雪崩
4. 异常分类：区分可重试和不可重试异常
5. 监控集成：支持各种事件监听和指标收集
6. 线程安全：使用数据库锁和Redis避免并发冲突