Laravel队列事件监听器模型查询偶发无报错终止问题排查

我正在开发一个作为Nuxt前端API服务的Laravel 10项目，需要处理数千个用于聚合报表的事件，要更新三个模型的聚合字段：

• AffiliateMinutelySummaryStat
• AffiliateHourlySummaryStat
• AffiliateDailySummaryStat

这三个模型的迁移文件基本一致，仅logged_at datetime字段的时间精度不同（分别精确到分、时、日）。

目前遇到的问题：队列事件监听器StoreAffiliateSummaryStat会在firstOrNew方法处偶发无报错终止执行。firstOrNew前的日志每次都会输出，但后续日志有时完全不触发，异常捕获逻辑也没抓到任何错误，导致模型字段无法完成增量更新。

项目使用Horizon + Redis作为队列服务，相关代码如下：

监听器代码 #

<?php

namespace App\Listeners;

use App\Events\AffiliateSummaryStatCreation;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Support\Facades\Log;
use App\Models\Reports\Minutely\AffiliateMinutelySummaryStat;
use App\Models\Reports\Hourly\AffiliateHourlySummaryStat;
use App\Models\Reports\Daily\AffiliateDailySummaryStat;
use Illuminate\Support\Sleep;
use Carbon\Carbon;

class StoreAffiliateSummaryStat implements ShouldQueue
{
    /**
     * The name of the connection the job should be sent to.
     *
     * @var string|null
     */
    public $connection = 'redis-short-running';

    /**
     * The name of the queue the job should be sent to.
     *
     * @var string|null
     */
    public $queue = 'statistics';

    /**
     * The number of times the queued listener may be attempted.
     *
     * @var int
     */
    public $tries = 1;

    /**
     * Create the event listener.
     */
    public function __construct()
    {
        //
    }

    /**
     * Get the tags that should be assigned to the job.
     *
     * @return array<int, string>
     */
    public function tags(): array
    {
        return ['affiliateStatistic'];
    }

    /**
     * Create statistics for each model
     */
    protected function createStatistic($model, $event)
    {
        try {
            $modelable = $model['model'];

            Sleep::for(rand(0, 3000))->milliseconds();

             Log::debug('always runs...');

            $summaryStatistic = $modelable::firstOrNew([
                'user_id' => isset($event->userId) && !empty($event->userId) ? $event->userId : null,
                'company_id' => isset($event->companyId) && !empty($event->companyId) ? $event->companyId : null,
                'affiliate_id' => isset($event->affiliateId) && !empty($event->affiliateId) ? $event->affiliateId : null,
                'brand' => isset($event->brand) && !empty($event->brand) ? $event->brand : null,
                'logged_at' => Carbon::parse($event->loggedAt)->format($model['format'])
            ]);

            $field = $event->column;

            Log::debug('sometimes never gets here????');

            $summaryStatistic->$field = $summaryStatistic->$field + $event->value;

            if ($field == 'total_processing_duration') {
                if (
                    $summaryStatistic->total_processing_duration > 0 &&
                    $summaryStatistic->total_leads > 0
                ) {
                    $summaryStatistic->avg_processing_duration = round(
                        $summaryStatistic->total_processing_duration /
                        $summaryStatistic->total_leads
                    );
                }
            }

            if ($field == 'total_processing_accept_duration') {
                if (
                    $summaryStatistic->total_processing_accept_duration > 0 &&
                    $summaryStatistic->total_accepted > 0
                ) {
                    $summaryStatistic->avg_processing_accept_duration = round(
                        $summaryStatistic->total_processing_accept_duration /
                        $summaryStatistic->total_accepted
                    );
                }
            }

            if ($field == 'total_processing_decline_duration') {
                if (
                    $summaryStatistic->total_processing_decline_duration > 0 &&
                    $summaryStatistic->total_declined > 0
                ) {
                    $summaryStatistic->avg_processing_decline_duration = round(
                        $summaryStatistic->total_processing_decline_duration /
                        $summaryStatistic->total_declined
                    );
                }
            }

            $summaryStatistic->save();

        } catch (\Exception $e) {
            Log::debug('No errors are thrown');
        }
    }

    /**
     * Handle the event.
     */
    public function handle(AffiliateSummaryStatCreation $event): void
    {
        $models = [[
            'model' => 'App\Models\Reports\Minutely\AffiliateMinutelySummaryStat',
            'format' => 'Y-m-d H:i:00'
        ], [
            'model' => 'App\Models\Reports\Hourly\AffiliateHourlySummaryStat',
            'format' => 'Y-m-d H:00:00'
        ], [
            'model' => 'App\Models\Reports\Daily\AffiliateDailySummaryStat',
            'format' => 'Y-m-d 23:59:59'
        ]];

        foreach ($models as $key => $model) {
            $this->createStatistic($model, $event);
        }
    }
}

事件分发示例 #

AffiliateSummaryStatCreation::dispatch(
    Carbon::now(),
    'total_submits',
    isset($affiliate) ? $affiliate->user_id : null,
    isset($affiliate) ? $affiliate->company_id : null,
    isset($affiliate) ? $affiliate->id : null,
);

1. 队列任务被外部终止（超时/内存耗尽） #

当前tries设为1，一旦任务因队列超时限制、PHP内存耗尽被强制终止，不会触发重试，也不会进入代码中的异常捕获逻辑（因为是外部进程终止）。

• 检查Horizon的任务超时配置，以及队列worker启动时的--timeout参数，适当调高超时时间；同时调整PHP的memory_limit值，避免内存不足。
• 将tries改为3以上，给任务重试机会，同时通过Horizon面板查看失败任务的详细原因（比如超时、内存错误）。

2. 数据库并发死锁 #

高并发场景下，多个队列进程同时操作同一统计行（比如同一时间维度的记录），会触发数据库行锁竞争，导致进程被直接终止，不会抛出PHP异常。

• 改用原子操作替代firstOrNew+save，避免行锁问题：

  $queryConditions = [
      'user_id' => $event->userId ?? null,
      'company_id' => $event->companyId ?? null,
      'affiliate_id' => $event->affiliateId ?? null,
      'brand' => $event->brand ?? null,
      'logged_at' => Carbon::parse($event->loggedAt)->format($model['format'])
  ];
  
  // 先确保记录存在，再原子增量
  $modelable::updateOrInsert($queryConditions, [$event->column => $event->value]);
  $modelable::where($queryConditions)->increment($event->column, $event->value);
  

• 或者使用lockForUpdate()手动加锁，确保同一时间只有一个进程操作该行：

  $summaryStatistic = $modelable::where($queryConditions)->lockForUpdate()->firstOrNew($queryConditions);
  

3. 异常捕获范围不足 #

代码中仅捕获\Exception，但PHP的致命错误、内存错误等属于\Error类，不会被\Exception捕获，导致错误信息丢失。

• 修改捕获逻辑为捕获\Throwable，同时输出详细错误信息：

  catch (\Throwable $e) {
      Log::error('统计任务执行失败', [
          'message' => $e->getMessage(),
          'trace' => $e->getTraceAsString(),
          'model' => $model['model'],
          'event_data' => (array)$event
      ]);
  }
  

4. 事件数据序列化问题 #

AffiliateSummaryStatCreation事件中的loggedAt是Carbon实例，队列序列化/反序列化后可能出现解析错误，导致Carbon::parse($event->loggedAt)抛出错误，但因捕获范围问题未被记录。

• 确保事件类的属性都是可序列化的，或者在事件类中实现Illuminate\Contracts\Queue\ShouldBeEncrypted接口，或自定义序列化逻辑。

5. 随机Sleep的潜在风险 #

Sleep::for(rand(0, 3000))的随机等待最长3秒，当队列积压时，可能导致任务超时被终止。

• 改为固定短时间等待（比如50毫秒），或者用数据库锁的方式替代这种并发规避手段。

内容的提问来源于stack exchange，提问作者Ryan H