Laravel 5.2 中的独特作业答案

【问题标题】：Unique Jobs in Laravel 5.2Laravel 5.2 中的独特作业
【发布时间】：2022-01-12 13:25:44
【问题描述】：

我在 laravel 5.2 中有一个遗留应用程序，我们使用队列来处理作业。我们观察到队列有时会获得过多的作业，这会导致重复作业被分派，因为之前的作业未完成处理，并且分派此类作业的 cron 再次运行并最终一遍又一遍地分派它们。

一个简单的解决方案是使这些工作独一无二，如果它是 laravel 8，这将是一个非常简单的改变。但是，我们处于 laravel 5.2 领域，所以我必须自己实施独特的工作。或者是否有人可以提出更好的选择？

另外，如果您要自己实施独特的工作，您会怎么做？我正在考虑的方法是：

将作业的唯一键添加到缓存或数据库表中（意味着获得了锁）处理完作业后清除条目（释放锁定）在调度作业之前，检查缓存中是否存在key（可以获取锁）

【问题讨论】：

这真的很模糊。例如，在我天真的头脑中，如果您的 cron 不会分派以前完成的作业，您是否不能将其修改为不分派以前分派的作业？我觉得可能会出现很多这样的问题，因为除了模糊的描述之外，这里没有关于你的过程的真正细节
在调度作业时使用withoutoverlapping 和onQueue 选项。
您在上一部分中描述了一个很好的解决方案。锁（又名互斥锁）适用于此用例。只有一件事，我也会检查工作内部的锁。由于在调度和作业执行之间可能有一些时间，我认为它更可靠。
@SachinKumar laravel 5.2 没有 withoutoverlapping 选项
所以他们可能会选择将工作放在队列中！

标签： php laravel laravel-5 caching locking

【解决方案1】：

根据 cmets 下的讨论和互联网上的更多帮助，我继续进行以下操作：

为作业创建唯一 ID
使用唯一 id 获取锁并将其存储在缓存中
如果已获得锁，则分派作业
作业完成后，从缓存中释放锁

可能不是最好的实现，但绝对符合我们的要求。

代码：

在调度作业之前获取锁的常用辅助函数：

/**
 * Acquires a lock but storing a key on the cache for the provided duration
 * the `$jobInstance` must have a `jobId()` function that provides a unique id
 * for the job used to prevent duplicacy
 *
 * @var  Job  $jobInstance  a Job instance to acquire a lock for
 * @var  int  $minutes  number of minutes to hold the lock for (default: 1 day)
 *
 * @return  bool
 */
function acquireLock($jobInstance, $minutes = 1440)
{
    $lockAcquired = false;
    try {
        DB::beginTransaction();
        /**
         * If the job instance does not provide a `jobId` function do not
         * attempt to acquire a lock for it
         */
        $jobId = method_exists($jobInstance, 'jobId') ? $jobInstance->jobId() : null;
        if (!is_null($jobId)) {
            $isLockAvailable = is_null(Cache::get($jobId));
            if ($isLockAvailable) {
                Cache::put($jobId, true, $minutes);
                $lockAcquired = true;
            }
        }
        DB::commit();
    } catch (\Throwable $th) {
        DB::rollback();
        Log::error("Unable to acquire lock");
        Log::error($th);
    }
    return $lockAcquired;
}

/**
 * Attempts to acquires a lock for the job and dispatches it
 * if the lock was successfull acquired
 *
 * @var  Job  $jobInstance  Job instance to dispatch
 *
 * @return void
 */
function dispatchWithLock($jobInstance)
{
    $isLockAcquired = acquireLock($jobInstance);
    if ($isLockAcquired) {
        dispatch($jobInstance);
    }
}

在AppServiceProvider 上添加了一个监听器来释放锁

Queue::after(function (JobProcessed $event) {
    /**
     * Check if the job has a `jobId` function and release the lock used to
     * maintain job uniqueness
     */
    $jobInstance = unserialize($event->data['data']['command']);
    $jobId = method_exists($jobInstance, 'jobId')
            ? $jobInstance->jobId()
            : null;
    if (!is_null($jobId)) {
        Cache::forget($jobId);
    }
});

为需要唯一的作业添加了jobId 函数

/**
 * Get a unique id for the job to prevent duplicates
 *
 * @return str
 */
public function jobId()
{
    return str_slug(get_class($this)) . $this->myModel->id;
}

最后用辅助函数调度作业

dispatchWithLock(new myJob($this));

【讨论】：