本文目录导读：
我来设计一个完善的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
]);
}
}
这个设计提供了完整的任务重试机制,包括：
- 灵活的配置：支持自定义重试次数、延迟策略、重试条件
- 持久化存储：数据库存储任务状态，支持分布式处理
- 指数退避：智能延迟算法避免雪崩
- 异常分类：区分可重试和不可重试异常
- 监控集成：支持各种事件监听和指标收集
- 线程安全：使用数据库锁和Redis避免并发冲突
