IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

PHP项目Laravel事件监听失败重试

wen PHP项目 3

本文目录导读：

PHP项目Laravel事件监听失败重试

  1. 方案一：使用队列 + 重试机制
  2. 方案二：自定义重试中间件
  3. 方案三：事件系统结合缓存实现重试
  4. 方案四：使用第三方包
  5. 方案五：完整的失败重试系统
  6. 最佳实践建议

在 Laravel 中实现事件监听失败重试机制,有几种方案：

使用队列 + 重试机制

配置队列重试

// config/queue.php
'connections' => [
    'redis' => [
        'driver' => 'redis',
        'connection' => 'default',
        'queue' => env('REDIS_QUEUE', 'default'),
        'retry_after' => 90,          // 重试延迟时间
        'block_for' => null,
        'after_commit' => null,
    ],
],

定义事件和监听器

// app/Events/OrderCreated.php
namespace App\Events;
use Illuminate\Foundation\Events\Dispatchable;
use Illuminate\Queue\SerializesModels;
class OrderCreated
{
    use Dispatchable, SerializesModels;
    public $order;
    public function __construct($order)
    {
        $this->order = $order;
    }
}
// app/Listeners/SendOrderNotification.php
namespace App\Listeners;
use App\Events\OrderCreated;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Queue\InteractsWithQueue;
class SendOrderNotification implements ShouldQueue
{
    use InteractsWithQueue;
    /**
     * 失败重试次数
     */
    public $tries = 3;
    /**
     * 最大执行时间（秒）
     */
    public $timeout = 60;
    /**
     * 重试延迟时间（秒）
     */
    public $backoff = [2, 5, 10]; // 分别对应第1、2、3次重试的延迟
    public function handle(OrderCreated $event)
    {
        // 业务逻辑
        try {
            // 发送通知
            $result = $this->sendNotification($event->order);
            if (!$result) {
                throw new \Exception('通知发送失败');
            }
        } catch (\Exception $e) {
            // 记录日志
            \Log::error('通知发送失败', [
                'order_id' => $event->order->id,
                'error' => $e->getMessage()
            ]);
            // 重新抛出异常以触发重试
            throw $e;
        }
    }
    private function sendNotification($order)
    {
        // 实际发送逻辑
        return true;
    }
    /**
     * 失败处理方法
     */
    public function failed(OrderCreated $event, $exception)
    {
        \Log::critical('通知发送最终失败', [
            'order_id' => $event->order->id,
            'error' => $exception->getMessage()
        ]);
    }
}

自定义重试中间件

// app/Queue/Middleware/CustomRetryMiddleware.php
namespace App\Queue\Middleware;
use Closure;
use Illuminate\Queue\Jobs\Job;
use Illuminate\Queue\Middleware\RateLimited;
class CustomRetryMiddleware
{
    /**
     * @param Job $job
     * @param Closure $next
     * @return mixed
     */
    public function handle($job, $next)
    {
        try {
            return $next($job);
        } catch (\Throwable $e) {
            // 判断是否需要重试
            if ($this->shouldRetry($job, $e)) {
                // 记录重试日志
                \Log::warning('任务执行失败，准备重试', [
                    'job' => get_class($job),
                    'attempt' => $job->attempts(),
                    'error' => $e->getMessage()
                ]);
                // 延迟重试
                $job->release(10); // 延迟10秒后重试
            } else {
                throw $e;
            }
        }
    }
    private function shouldRetry($job, $exception)
    {
        // 重试次数限制
        $maxAttempts = 3;
        if ($job->attempts() >= $maxAttempts) {
            \Log::error('任务达到最大重试次数', [
                'job' => get_class($job),
                'attempts' => $job->attempts()
            ]);
            return false;
        }
        // 可以基于异常类型判断是否重试
        if ($exception instanceof \Illuminate\Database\QueryException) {
            // 数据库异常，不重试
            return false;
        }
        return true;
    }
}

事件系统结合缓存实现重试

// app/Listeners/FailedEventListener.php
namespace App\Listeners;
use Illuminate\Cache\RedisStore;
use Illuminate\Support\Facades\Cache;
class FailedEventListener
{
    /**
     * 处理事件
     */
    public function handle($event)
    {
        // 存储处理状态
        $cacheKey = 'event_retry_' . get_class($event) . '_' . $event->id;
        if (!Cache::has($cacheKey)) {
            Cache::put($cacheKey, ['status' => 'processing'], 3600);
        }
        try {
            // 业务逻辑
            $result = $this->processEvent($event);
            // 处理成功，存储状态
            Cache::put($cacheKey, ['status' => 'success'], 3600);
            return $result;
        } catch (\Exception $e) {
            // 记录失败信息
            $failed = Cache::get($cacheKey);
            $attempts = isset($failed['attempts']) ? $failed['attempts'] : 0;
            $attempts++;
            if ($attempts < 3) {
                // 设置下次重试时间
                $retryTime = now()->addMinutes(5);
                Cache::put($cacheKey, [
                    'status' => 'failed',
                    'attempts' => $attempts,
                    'next_retry' => $retryTime,
                    'last_error' => $e->getMessage()
                ], 3600);
                // 调度重试任务
                $this->scheduleRetry(get_class($event), $event->id, $retryTime);
            } else {
                \Log::critical('事件处理多次失败', [
                    'event' => get_class($event),
                    'id' => $event->id,
                    'attempts' => $attempts
                ]);
            }
            throw $e;
        }
    }
    private function scheduleRetry($eventClass, $eventId, $retryTime)
    {
        // 使用 Laravel 的调度器
        $schedule = app(\Illuminate\Console\Scheduling\Schedule::class);
        $schedule->call(function () use ($eventClass, $eventId) {
            $event = $eventClass::find($eventId);
            if ($event) {
                event(new $eventClass($event));
            }
        })->at($retryTime->format('Y-m-d H:i'));
    }
}

使用第三方包

安装 Laravel Horizon 支持队列监控

composer require laravel/horizon
php artisan horizon:install
php artisan migrate

Horizon 配置

// config/horizon.php
'defaults' => [
    'supervisor-1' => [
        'connection' => 'redis',
        'queue' => ['default'],
        'balance' => 'auto',
        'processes' => 1,
        'tries' => 3,              // 重试次数
        'timeout' => 60,           // 超时时间
        'nice' => 0,
    ],
],

完整的失败重试系统

// app/Services/EventRetryService.php
namespace App\Services;
use Illuminate\Support\Facades\Redis;
use Illuminate\Support\Facades\Log;
class EventRetryService
{
    protected $maxRetries = 3;
    protected $retryDelays = [1, 5, 15]; // 分钟
    /**
     * 注册事件监听器
     */
    public function registerListeners(array $events)
    {
        foreach ($events as $eventClass => $listenerClass) {
            \Event::listen($eventClass, $listenerClass);
        }
    }
    /**
     * 处理事件失败
     */
    public function handleFailure($event, $exception, $listenerClass)
    {
        $eventId = $this->getEventId($event);
        $retryKey = "event_retry:{$eventId}";
        // 获取当前重试次数
        $attempts = (int) Redis::hget($retryKey, 'attempts');
        $attempts++;
        // 记录失败信息
        Redis::hset($retryKey, 'attempts', $attempts);
        Redis::hset($retryKey, 'last_error', $exception->getMessage());
        Redis::hset($retryKey, 'listener', $listenerClass);
        // 判断是否继续重试
        if ($attempts <= $this->maxRetries) {
            $this->scheduleRetry($event, $attempts);
        } else {
            $this->markAsPermanentFailure($eventId);
        }
        // 记录日志
        Log::error('事件处理失败', [
            'event' => get_class($event),
            'listener' => $listenerClass,
            'attempts' => $attempts,
            'error' => $exception->getMessage()
        ]);
    }
    /**
     * 调度重试
     */
    protected function scheduleRetry($event, $attempts)
    {
        $delay = $this->retryDelays[$attempts - 1] ?? 30; // 分钟
        // 使用队列延迟执行
        \Queue::later(
            now()->addMinutes($delay),
            new \App\Jobs\RetryEventJob($event)
        );
        // 或者使用 Redis 的延迟队列
        Redis::zadd('event_retry_queue', 
            time() + ($delay * 60), 
            serialize($event)
        );
    }
    /**
     * 标记永久失败
     */
    protected function markAsPermanentFailure($eventId)
    {
        $key = "event_retry:{$eventId}";
        Redis::hset($key, 'status', 'permanent_failure');
        Log::critical("事件处理永久失败，事件ID: {$eventId}");
    }
}
// app/Jobs/RetryEventJob.php
namespace App\Jobs;
use Illuminate\Bus\Queueable;
use Illuminate\Queue\SerializesModels;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
class RetryEventJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
    protected $event;
    public function __construct($event)
    {
        $this->event = $event;
    }
    public function handle()
    {
        try {
            // 重新触发事件
            event($this->event);
        } catch (\Exception $e) {
            // 如果还是失败，记录日志
            Log::error('重试事件失败', [
                'event' => get_class($this->event),
                'error' => $e->getMessage()
            ]);
            // 可以考虑再次触发重试
            if ($this->attempts() < 3) {
                $this->release(60 * 5); // 5分钟后重试
            }
        }
    }
}

最佳实践建议

  1. 设置合理的重试次数：不要无限重试，一般 3-5 次即可
  2. 使用指数退避：逐步增加重试间隔
  3. 区分错误类型：
    • 可重试错误（网络问题、外部服务暂时不可用）
    • 不可重试错误（数据验证错误、权限问题）
  4. 监控通知：设置告警机制
  5. 记录完整日志：便于排查问题
// 注册事件时指定重试策略
Event::listen(
    OrderCreated::class,
    [SendOrderNotification::class, 'handle'],
    10 // 优先级
);
// 监听器类定义重试参数
class SendOrderNotification implements ShouldQueue
{
    public $tries = 3;
    public $timeout = 60;
    public $backoff = [1, 5, 10];
    public $maxExceptions = 3;
}

选择哪种方案取决于你的具体需求：

  • 简单场景：使用方案一（队列配置即可）
  • 复杂业务：使用方案二（自定义中间件）
  • 需要监控：使用方案四（Horizon）
  • 分布式系统：使用方案三（缓存/Redis）

上一篇PHP项目Laravel日志上下文额外数据

下一篇PHP项目Laravel异常处理区分环境

相关文章

抱歉，评论功能暂时关闭!