php - 检查 Queue::before 中的 Laravel Jobs 以在处理之前删除

Question

我得到了使用 Queue::before 查看 AppServiceProvider 的想法，以此来为我不再想运行的作业添加检查并删除它们，而不必为我编写的每个作业添加检查。
背景，我正在开发一个进行审计的 SaaS，因此审计可以运行数小时并成为 1000 个作业。如果我可以在作业完成时在作业中查找审核 ID 并将其与已取消的任何审核 ID 的缓存数组进行比较，我就可以节省时间。

所以我要做的是如何在 Queue::before 中解包 Job 以获取要检查的 ID？ (正常的 laravel 队列代码，并使用 RabbitMQ)

由于作业被包裹在一层或两层事件类中，我无法将数据转储到屏幕上查看，只能转储到日志文件中，因为它在队列中。

在 app/Providers/AppServiceProvider.php 中:

Queue::before(function (JobProcessing $event) {
    // $event->connectionName
    // $event->job
    $job = $event->job->payload();
    $obj = unserialize($job['data']['data']);
}

就我感兴趣的事件而言，有效载荷有数据，有数据，这是我感兴趣的序列化对象。这似乎不是最好的方法，或者看看如何与之交互以更好的方式。

谢谢

Answer 1

我正处于涉及 webhook 交付的类似问题中。通过开发人员门户，我们允许用户重新排队 webhook(以缩短对退回交付尝试的等待)。由于这可能会为同一个 webhook 创建第二个作业，因此我们想方设法将原始作业标识为已过期。

app/Jobs/DeliverWebhook.php 构造函数:

public function __construct(Webhook $webhook)
{
    $this->webhook      = $webhook;
    $this->queued_at    = Carbon::now();

    Cache::put(
        'DeliverWebhook.'. $this->webhook->id .'.QueuedAt',
        $this->queued_at,
        Carbon::now()->addDays(3)
    );
}

在这里，您可以看到我们已将 queued_at 属性附加到此作业实例。 (我们也可以通过使用 uniqid() 或 random_bytes() 之类的东西来使它更独特，以避免排队时潜在的双击问题或类似的问题。)

第二部分是我们设置半唯一缓存键来匹配这个queued_at时间。我将其设置为 3 天后过期，即在我们的后退重试尝试结束后。

现在，当作业被拿起处理以进行处理时，我可以检查作业实例的 queued_at 属性属性与缓存值一起删除，并删除该作业。

在我的 AppServiceProvider boot 方法中:

Queue::before(function ($event) {
    if ($event->job->queue == 'webhooks' && $event->job->getName() == 'DeliverWebhook') {
        $cache_key = 'DeliverWebhook.'. $event->job->instance->webhook->id .'QueuedAt';

        if ($event->job->instance->queued_at < Cache::get($cache_key)) {
            $event->job->delete();

            throw new JobRequeuedException;
        }
    }
});

最后会抛出一个异常，因为默认情况下，queue worker 在调用$job->fire() 之前不会检查作业是否被删除。抛出异常会强制工作人员跳过 fire() 并跳转到 handleJobException() 方法。

注意:我仍然需要对此进行适当的测试。