I have a legacy application in laravel 5.2 and we use queues to process jobs. We observed that the queue can sometimes get too many jobs which leads to duplicate jobs getting dispatched because the previous jobs don't complete processing and the cron that dispatches such jobs runs again and ends up dispatching them over and over.

A simple solution would be to make these jobs unique which would be a very simple change if it were laravel 8. However, we're in laravel 5.2 territory so i'll have to implement unique jobs myself. Or if someone could suggest a better alternative?

Also, if you were to implement unique jobs yourself, how would you do it? The approach I'm thinking is:

Add a unique key for the job to the cache or a database table (implying a lock is attained) Clear the entry once the job is processed (lock released) Before dispatching the job, check if the key exists in the cache or not (lock can be attained or not)

CodePudding user response：

Based on the discussions under comments and some more help from the internet, I went ahead with the following:

1. Create a unique id for the jobs
2. Acquire a lock using the unique id and store it in the cache
3. If the lock was acquired, dispatch the job
4. After the job's completion, release the lock from the cache

May not be the best implementation but definitely meets our requirement.

Code:

Common helper functions to acquire a lock before dispatching jobs:

/**
 * 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);
    }
}

Added a listener on AppServiceProvider to release the locks

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);
    }
});

Added a jobId function to jobs that need to be unique

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

Finally dispatch jobs with the helper function

dispatchWithLock(new myJob($this));